Реализация рассуждений и планирования на Python
В этой главе рассматриваются ключевые интеллектуальные возможности AI Agent, а именно способности к рассуждению и планированию.
В отличие от традиционных моделей выполнения инструкций, Agent с возможностями рассуждения может осуществлять сложные мыслительные процессы.
Он может декомпозировать сложные задачи на выполнимые шаги и принимать решения на основе контекста.
Фреймворк ReAct
ReAct (рассуждение + действие) — это фреймворк, объединяющий рассуждение и действие.
В паттерне ReAct Agent поочерёдно переходит от рассуждения к действию, образуя замкнутый цикл.
Рассуждение направляет действие, а результаты действий возвращаются в процесс рассуждения, повторяясь в этом цикле.
ReAct имитирует процесс решения задач человеком: обдумать, прежде чем действовать, наблюдать результаты после действия, а затем продолжать рассуждать на основе новой информации.
Принцип работы
Основную идею ReAct можно свести к циклу из трёх фаз:
Во-первых, фаза рассуждения. Агент формирует мысль о следующем действии на основе текущего контекста.
Во-вторых, фаза действия. Агент выполняет выбранный инструмент или действие.
В-третьих, фаза наблюдения. Агент получает результат действия, обновляет контекст и возвращается к первому шагу, чтобы продолжить рассуждать.
Реализация кода
Ниже приведена упрощённая реализация ReAct-агента:
Базовая реализация ReAct-агента
"""
Реализация агента на основе фреймворка ReAct
Основной механизм: рассуждение -> действие -> наблюдение -> рассуждение (цикл)
"""
def __init__(self, llm, tools, max_iterations=5):
# Экземпляр LLM, используемый для рассуждений и генерации
self.llm = llm
# Список доступных инструментов
self.tools = tools
# Максимальное количество итераций для предотвращения бесконечных циклов
self.max_iterations = max_iterations
def run(self, task):
"""
Выполнить цикл ReAct
:param task: Описание задачи пользователя
:return: Результат выполнения
"""
# Инициализировать контекст, включая задачу и информацию об истории
context = {
"task": task,
"steps": [], # Запись выполненных шагов
"observations": [] # Запись результатов наблюдений
}
for i in range(self.max_iterations):
# Этап 1: Рассуждение — сгенерировать следующее действие на основе текущего контекста
reasoning = self.llm.reason(context)
# Определить, нужно ли выполнять инструмент, или вернуть ответ напрямую
if reasoning.needs_action:
# Выбрать инструмент для использования
action = reasoning.select_tool(self.tools)
# Выполнить инструмент и получить результат наблюдения
observation = action.execute(reasoning.tool_input)
# Добавить результат наблюдения в контекст
context["observations"].append(observation)
else:
# Действие не требуется, напрямую вернуть ответ, полученный в результате рассуждения
return reasoning.answer
# Достигнуто максимальное число итераций без завершения
return "Достигнут максимальный лимит итераций"
Типичные сценарии применения
Шаблон ReAct особенно подходит для следующих сценариев:
Задачи, которые необходимо выполнить после исследования среды и сбора информации.
Сценарий интеллектуального поиска: Агенту сначала нужно найти информацию, затем на основе результатов поиска выполнить рассуждение и обобщение.
Диалоговые вопросы и ответы: Агенту необходимо уточнять требования и получать информацию посредством многоходового диалога.
Решение сложных задач: проблему нельзя решить за один шаг, требуется несколько промежуточных шагов.
Примечание: преимущество паттерна ReAct заключается в его гибкости, позволяющей динамически корректировать последующие действия на основе результата выполнения каждого шага. Но это также означает, что путь выполнения может быть нестабильным, что делает его подходящим для задач, требующих исследования.
Цепочка рассуждений (CoT)
Цепочка рассуждений — это техника, которая побуждает модель демонстрировать пошаговый процесс рассуждений.
CoT не требует от модели сразу давать ответ, а направляет её сначала показать шаги рассуждения, а затем прийти к окончательному выводу.
Почему нужна цепочка рассуждений
Прямое требование к модели выдать ответ имеет несколько проблем:
Промежуточные шаги в сложных рассуждениях легко упускаются или пропускаются.
Трудно точно определить, где возникла ошибка.
Пользователи не могут понять процесс мышления модели.
Chain-of-Thought решает эти проблемы, требуя от модели показывать процесс своих рассуждений.
Zero-shot CoT
Zero-shot CoT — это метод, который позволяет вызвать способность к пошаговому рассуждению без необходимости в примерах.
Просто добавьте триггерную фразу вроде "Давайте думать шаг за шагом" в конец промпта.
Пример Zero-shot CoT
prompt = """
Вопрос: У Сяо Мина 5 яблок. Сяо Хун даёт ему 3. Сяо Мин съедает 2. Сколько осталось?
Давайте думать шаг за шагом:
"""
# Вызов языковой модели
response = llm.generate(prompt)
# Пример вывода модели:
# Шаг 1: Вычислить общее количество яблок после того, как Сяо Мин их получит
# Изначально у Сяо Мина было 5 яблок, Сяо Хун дала ему 3
# 5 + 3 = 8
#
# Шаг 2: Вычислить количество после еды
# Сяо Мин съел 2
# 8 - 2 = 6
#
# Вывод: осталось 6 яблок
Few-shot CoT
Few-shot CoT помогает модели осваивать определённые паттерны рассуждений, предоставляя примеры, которые включают подробные процессы рассуждения.
Когда Zero-shot CoT неэффективен, можно попробовать Few-shot CoT.
Пример Few-shot CoT
prompt = """
Пример 1:
Вопрос: У Сяо Чжана есть 10 юаней. Он покупает 3 книги по 2 юаня каждая. Сколько осталось?
Давайте думать шаг за шагом:
- У Сяо Чжана изначально было 10 юаней
- Каждая книга стоит 2 юаня, покупает 3 книги, тратит 3 × 2 = 6 юаней
- 10 - 6 = 4 юаня
Ответ: осталось 4 юаня
Пример 2:
Вопрос: Кошка ловит 2 мыши в час. Сколько она поймает за 8 часов?
Давайте думать шаг за шагом:
- Ловит 2 мыши в час
- За 8 часов ловит 8 × 2 = 16
Ответ: поймала 16
Вопрос: {user_question}
Давайте думать шаг за шагом:
"""
response = llm.generate(prompt)
Преимущества цепочки рассуждений
Технология цепочки рассуждений приносит три ключевые ценности:
Прослеживаемость: разбивает сложные рассуждения на прослеживаемые промежуточные шаги, делая процесс рассуждения более понятным.
Интерпретируемость: повышает интерпретируемость модели, позволяя пользователям понимать, откуда берутся ответы.
Повышение точности: значительно повышает точность на сложных задачах рассуждения, особенно эффективно в математических и логических задачах.
Дерево мыслей（ToT）
Дерево мыслей — это расширение цепочки мыслей; оно больше не ограничивается линейным рассуждением.
ToT исследует несколько возможных путей в каждом узле рассуждения, образуя древовидную структуру.
Это позволяет агенту выполнять многопутевое исследование, возврат назад и глобальную оценку.
Отличия от цепочки мыслей
CoT использует линейное рассуждение, где каждый шаг зависит от вывода предыдущего шага, что подходит для задач с ясными путями.
ToT использует пространственное рассуждение, рассматривая несколько возможных ветвей одновременно, что подходит для задач, требующих исследования и планирования.
Реализация кода
Базовая реализация ToT-агента
"""
Реализация агента «Дерево мыслей»
Основной механизм: сгенерировать несколько ветвей-кандидатов в каждом узле, оценить их, затем выбрать лучшую для продолжения.
"""
def __init__(self, llm, max_depth=4, beam_size=3):
# Экземпляр LLM
self.llm = llm
# Максимальная глубина поиска
self.max_depth = max_depth
# Количество узлов-кандидатов, сохраняемых на каждом уровне (ширина луча)
self.beam_size = beam_size
def solve(self, problem):
"""
Решение задач с использованием фреймворка ToT
:param problem: Описание задачи
:return: Оптимальное решение
"""
# Создать корневой узел, содержащий задачу как начальное состояние
root = ThoughtNode(problem, depth=0)
# Инициализировать список узлов фронтира (узлов для раскрытия)
frontier = [root]
# Раскрывать слой за слоем
for depth in range(self.max_depth):
# Сохранить все узлы-кандидаты
all_candidates = []
# Перебрать все узлы фронтира
for node in frontier:
# Сгенерировать несколько возможных следующих шагов для текущего узла
candidates = self.llm.generate_thoughts(
node.content, # Содержимое текущего узла
n=self.beam_size # Количество генераций
)
# Создать новый узел и добавить его в список кандидатов
for cand in candidates:
all_candidates.append(
ThoughtNode(cand, depth + 1, parent=node)
)
# Оценить все узлы-кандидаты
evaluated = self.evaluator.rank(all_candidates)
# Выбрать beam_size лучших узлов в качестве следующего фронта
frontier = evaluated[:self.beam_size]
# Проверить, найдено ли решение
if self.is_solution(frontier):
break
# Выполнить обратный поиск, чтобы найти оптимальное решение
return self.backtrack_best(frontier)
Сценарии применения
ToT особенно подходит для сценариев, требующих выбора или планирования:
Творческое письмо: создайте несколько направлений развития сюжета, оцените их, затем выберите лучшее.
Стратегическое планирование: оцените потенциальные результаты нескольких планов действий.
Сложное принятие решений: задачи принятия решений, требующие рассмотрения нескольких возможностей.
Планирование задач и MCTS
Поиск по дереву Монте-Карло (MCTS) — это эвристический алгоритм поиска, используемый для сложных задач принятия решений.
MCTS широко применяется в планировании агентов, особенно подходит для игрового ИИ и сложного планирования задач.
Ключевая идея
MCTS оценивает потенциальную ценность каждого узла решения путём моделирования случайных игр.
Ему не нужно оценивать все возможные пути, вместо этого он использует выборку и статистику, чтобы направлять поиск.
Четыре шага MCTS
Шаг 1, Выбор: начиная с корневого узла, рекурсивно выбирайте оптимальный дочерний узел, пока не достигнете листового узла.
На этапе выбора используется формула UCB (Upper Confidence Bound) для балансировки исследования и использования.
Шаг 2, Расширение: добавьте один или несколько дочерних узлов к листовому узлу.
Шаг 3, Симуляция: начиная с нового узла, случайным образом симулируйте игру до конца.
Шаг 4, Обратное распространение: обновите статистику всех узлов вдоль пути симуляции.
Реализация кода
Реализация планировщика MCTS
class MCTSNode:
"""
Узел дерева MCTS
Хранит состояние узла, статистику и связи с дочерними узлами
"""
def __init__(self, state, parent=None, action=None):
# Текущее состояние
self.state = state
# Ссылка на родительский узел
self.parent = parent
# Действие от родительского узла к этому узлу
self.action = action
# Список дочерних узлов
self.children = []
# Количество посещений этого узла
self.visit_count = 0
# Суммарное значение вознаграждения этого узла
self.reward = 0.0
def is_fully_expanded(self):
"""Проверить, все ли возможные дочерние узлы раскрыты"""
return len(self.children) > 0
def is_terminal(self):
"""Проверить, является ли он терминальным узлом (игра окончена или цель достигнута)"""
return self.state.is_terminal()
def uct_child(self):
"""
Использовать формулу UCT для выбора оптимального дочернего узла
UCT = reward/visits + C * sqrt(ln(parent_visits)/visits)
C — константа исследования, обычно задаётся как sqrt(2)
"""
# Константа исследования, балансирующая исследование и эксплуатацию
C = math.sqrt(2)
return max(
self.children,
key=lambda c: c.reward / c.visit_count +
C * math.sqrt(math.log(self.visit_count) / c.visit_count)
)
class MCTSPlanner:
"""
Планировщик MCTS
Использует поиск по дереву Монте-Карло для генерации планов действий для Агента
"""
def __init__(self, simulation_limit=1000, exploration_constant=1.41):
# Максимальное количество симуляций
self.simulation_limit = simulation_limit
# Константа исследования
self.exploration_constant = exploration_constant
def plan(self, initial_state):
"""
Начать планирование из начального состояния
:param initial_state: начальное состояние
:return: оптимальное действие
"""
# Создать корневой узел
root = MCTSNode(initial_state)
# Выполнить несколько симуляций
for _ in range(self.simulation_limit):
# 1. Выбор: Выбрать оптимальный дочерний узел от корневого узла вниз, пока не будет достигнут листовой узел
node = self._selection(root)
# 2. Расширение: Если это не терминальный узел, расширить новый узел
if not node.is_terminal():
node = self._expansion(node)
# 3. Симуляция: случайное моделирование от нового узла до терминального узла
reward = self._simulation(node)
# 4. Обратное распространение: обновить статистику всех узлов на пути
self._backpropagation(node, reward)
# Вернуть действие, соответствующее оптимальному дочернему узлу корневого узла
return root.best_child().action
def _selection(self, node):
"""Фаза выбора: выбрать оптимальный дочерний узел"""
while node.is_fully_expanded() and not node.is_terminal():
node = node.uct_child()
return node
def _expansion(self, node):
"""Фаза расширения: добавить новый дочерний узел"""
# Сгенерировать все возможные действия
possible_actions = node.state.get_possible_actions()
# Создать дочерний узел для каждого действия
for action in possible_actions:
new_state = node.state.apply_action(action)
child = MCTSNode(new_state, parent=node, action=action)
node.children.append(child)
# Вернуть случайный дочерний узел (или использовать детерминированную стратегию)
return node.children[0]
def _simulation(self, node):
"""Фаза симуляции: случайным образом симулировать до завершения игры"""
state = node.state
while not state.is_terminal():
# Случайным образом выбрать действие
actions = state.get_possible_actions()
action = random.choice(actions)
state = state.apply_action(action)
# Вернуть итоговую награду
return state.get_reward()
def _backpropagation(self, node, reward):
"""Обратное распространение: обновление статистики"""
while node is not None:
node.visit_count += 1
node.reward += reward
node = node.parent
Примечание: MCTS требует больших вычислительных затрат и подходит для сценариев, требующих глубокого планирования, но с чёткими условиями завершения. Для задач с высокими требованиями к реальному времени может потребоваться ограничить количество симуляций или использовать другие методы.
Reflexion (саморефлексия)
Reflexion — это техника, которая даёт агенту способности к саморефлексии.
Добавив агенту механизм рефлексии, он может анализировать причины неудачи после неудачи, корректировать свою стратегию и повторять попытку.
Ключевая идея
Агент не только выполняет действия, но также наблюдает за результатами и размышляет: Почему я потерпел неудачу? Как мне улучшиться в следующий раз?
Эта способность имеет решающее значение для непрерывного обучения и самосовершенствования.
Реализация кода
Реализация Reflexion-агента
"""
Агент со способностью к саморефлексии
Основной механизм: Выполнить -> Оценить -> Отрефлексировать -> Повторить
"""
def __init__(self, actor, reviewer, max_retries=3):
# Исполнитель: отвечает за выполнение конкретных задач
self.actor = actor
# Оценщик: отвечает за оценку результатов выполнения
self.reviewer = reviewer
# Максимальное количество повторных попыток
self.max_retries = max_retries
def run(self, task):
"""
Выполнить задачу с саморефлексией
:param task: описание задачи
:return: результат выполнения
"""
# Ведение истории выполнения
history = []
for attempt in range(self.max_retries):
# Фаза 1: Попытка выполнить задачу
result = self.actor.execute(task, history)
# Записать эту попытку
history.append({
"attempt": attempt,
"result": result
})
# Фаза 2: Оценить результат
feedback = self.reviewer.evaluate(task, result)
# Фаза 3: Проверить, удалось ли это
if feedback.is_success:
return result
# Фаза 4: Поразмышлять о причине неудачи
# Сгенерировать новый стратегический промпт, чтобы направить следующую попытку
reflection = self.reviewer.reflect(
task, # Исходная задача
result, # Неудачный результат
feedback # Обратная связь по оценке
)
# Добавить результат рефлексии в историю для использования в следующей попытке
history.append({
"type": "reflection",
"content": reflection
})
# Если все повторные попытки не удались, вернуть историю
return history
class Reviewer:
"""Оценщик: оценивает результаты выполнения и генерирует рефлексии"""
def evaluate(self, task, result):
"""
Оценить результат выполнения
:return: объект, содержащий сведения об успешности и подробную обратную связь
"""
# Проверить, удовлетворяет ли результат требованиям задачи
is_success = self.check_success(task, result)
if is_success:
return EvaluationResult(is_success=True)
else:
# Сгенерировать анализ причин сбоя
failure_reasons = self.analyze_failures(task, result)
return EvaluationResult(
is_success=False,
reasons=failure_reasons
)
def reflect(self, task, result, feedback):
"""
Сгенерировать содержимое рефлексии
Помогите Actor понять причину неудачи и улучшить свою стратегию
"""
prompt = f"""
Задача: {task}
Результат выполнения: {result}
Причины неудачи: {feedback.reasons}
Пожалуйста, проанализируйте причины неудачи и предоставьте предложения по улучшению.
Ключевые моменты:
1. Что пошло не так?
2. Как избежать этого в следующий раз?
3. Какую стратегию нужно изменить?
"""
return self.llm.generate(prompt)
Сценарии применения
Reflexion особенно подходит для следующих сценариев:
Задачи, требующие постоянного улучшения, такие как диалоговые системы, генерация кода и т. д.
Сценарии, в которых стоимость ошибок высока, но стоимость повторной попытки низка.
Ситуации, в которых необходимо учиться на неудачах.
Стратегии декомпозиции задач
Сложные задачи обычно необходимо декомпозировать на управляемые подзадачи.
Эффективная декомпозиция задач — основа способности к планированию.
Рекурсивная декомпозиция задач
Рекурсивно декомпозировать задачу на более мелкие подзадачи, пока подзадачи не смогут быть выполнены напрямую.
Рекурсивная декомпозиция задач
"""
Рекурсивно декомпозировать задачи
:param task: Задача для декомпозиции
:param is_executable_fn: Функция для определения, можно ли выполнить задачу напрямую
:param decompose_fn: Функция для декомпозиции задачи
:return: Список исполняемых задач
"""
# Если задачу можно выполнить напрямую, вернуть её напрямую
if is_executable_fn(task):
return [task]
# Разбить задачу на подзадачи
subtasks = decompose_fn(task)
# Рекурсивно разбить каждую подзадачу
result = []
for subtask in subtasks:
# Рекурсивно вызвать функцию декомпозиции для подзадач
result.extend(
decompose(subtask, is_executable_fn, decompose_fn)
)
return result
# Пример: определить, является ли задача исполняемой
def is_code_complete(task):
"""Проверить, может ли задача напрямую выполнить код"""
return task.get("type") == "code" and len(task.get("dependencies", [])) == 0
# Пример: декомпозировать сложную задачу
def decompose_programming_task(task):
"""Декомпозировать задачу программирования"""
if "write_function" in task["type"]:
return [
{"type": "understand_requirements", "task": task["spec"]},
{"type": "write_code", "spec": task["spec"]},
{"type": "write_tests", "function": task["function_name"]},
{"type": "verify_tests", "function": task["function_name"]}
]
return [task]
Параллельная декомпозиция задач
Определите независимые подзадачи, которые можно выполнять параллельно, чтобы повысить эффективность выполнения.
Это ключевая стратегия ускорения выполнения задач.
Иерархическая декомпозиция задач
Разделяйте задачи на разные уровни абстракции; задачи высокого уровня вызывают задачи низкого уровня, образуя дерево иерархии задач.
Подходит для сложных систем, требующих нескольких уровней абстракции.
Паттерн Plan-and-Execute
Plan-and-Execute — это архитектурный паттерн, который отделяет планирование от выполнения.
Агент сначала полностью планирует весь поток задач, а затем выполняет его согласно плану.
Отличия от ReAct
ReAct чередует рассуждение и выполнение, что даёт большую гибкость, но путь может быть нестабильным.
Plan-and-Execute сначала планирует, а затем выполняет, что обеспечивает большую стабильность, но лишено возможности динамической корректировки.
Реализация кода
Реализация агента Plan-and-Execute
"""
Агент с архитектурой Plan-and-Execute
Основной механизм: сначала полностью спланировать, затем выполнять согласно плану
"""
def __init__(self, planner, executor):
# Планировщик: отвечает за создание плана выполнения
self.planner = planner
# Исполнитель: отвечает за выполнение конкретных шагов
self.executor = executor
def run(self, task):
"""
Выполнить задачу
Разделено на фазу планирования и фазу выполнения
"""
# ==================== Фаза планирования ====================
# Сгенерировать полный план выполнения за один раз
plan = self.planner.create_plan(task)
# ==================== Фаза выполнения ====================
results = []
# Выполнять по порядку согласно плану
for step in plan.steps:
# Выполнить текущий шаг
result = self.executor.execute(step)
results.append(result)
# Проверить, требуется ли перепланирование
# Например: результаты выполнения не соответствуют ожиданиям или возникают непредвиденные ситуации
if self.needs_replan(results):
# Перепланировать на основе текущих результатов
plan = self.planner.replan(task, results)
# Интегрировать все результаты
return self.summarize(results)
def needs_replan(self, results):
"""
Определить, требуется ли перепланирование
Проверьте, соответствуют ли результаты выполнения ожиданиям
"""
# Получить последний результат
last_result = results[-1]
# Если результат явно не соответствует ожиданиям, требуется перепланирование
if last_result.is_unexpected():
return True
# Если результат делает невозможным выполнение последующих шагов, требуется перепланирование
if last_result.blocks_future():
return True
return False
class Planner:
"""Planner: создать план выполнения задачи"""
def create_plan(self, task):
"""Создать первоначальный план выполнения"""
prompt = f"""
Задача: {task}
Пожалуйста, создайте подробный план выполнения, включая:
1. Порядок шагов для выполнения
2. Конкретные операции для каждого шага
3. Зависимости между шагами
Формат вывода:
Шаг 1: [Описание операции]
Шаг 2: [Описание операции]
...
"""
plan_text = self.llm.generate(prompt)
return Plan.parse(plan_text)
def replan(self, task, results):
"""Перепланирование на основе результатов выполнения"""
# Проанализировать выполненные результаты
completed = [r for r in results if r.is_success]
failed = [r for r in results if not r.is_success]
# Сгенерировать скорректированный план
prompt = f"""
Исходная задача: {task}
Выполненные шаги: {completed}
Неудавшиеся шаги: {failed}
Пожалуйста, создайте скорректированный план выполнения:
"""
plan_text = self.llm.generate(prompt)
return Plan.parse(plan_text)
class Executor:
"""Исполнитель: выполнить конкретные шаги плана"""
def execute(self, step):
"""Выполнить один шаг"""
# Выбрать метод выполнения на основе типа шага
if step.type == "tool_call":
return self.execute_tool_call(step)
elif step.type == "code":
return self.execute_code(step)
elif step.type == "query":
return self.execute_query(step)
return Result(success=False, error="Unknown step type")
Примечание: Режим Plan-and-Execute подходит для сценариев, где структура задачи относительно стабильна и может быть спланирована заранее. Для сред, требующих гибкой адаптации, режим ReAct может подойти больше.
Краткое содержание главы
В этой главе представлены основные возможности рассуждения и планирования AI-агентов.
ReActФреймворк позволяет агенту корректировать свою стратегию во время выполнения с помощью цикла рассуждения и действия.
Цепочка рассуждений (CoT)Повышает точность выполнения сложных задач за счёт демонстрации пошагового процесса рассуждения.
Дерево мыслей (ToT)Расширяет цепочку рассуждений, поддерживая многопутевое исследование и возврат назад.
MCTSЭто эвристический алгоритм поиска, подходящий для сложных задач принятия решений и планирования.
ReflexionНаделите агента возможностями саморефлексии, чтобы учиться на неудачах и совершенствоваться.
Стратегия декомпозиции задачЭто основа способности к планированию, включая рекурсивную декомпозицию, параллельную декомпозицию и иерархическую декомпозицию.
Plan-and-ExecuteИспользует подход «сначала планирование, затем выполнение», подходит для задач со стабильной структурой.
Эти технологии можно использовать по отдельности или в комбинации для создания более мощных агентных систем.
Другие расширения