Цель — научиться создавать модульные (multi-stage) системы на базе LLM, а затем оптимизировать промпты (инструкции и примеры) таким образом, чтобы итоговая метрика качества (accuracy, retrieval score и т.п.) превышала вариант с ручным подбором текста промпта.
Почему нужны многошаговые LM-программы и оптимизация промпта
Современные большие языковые модели (LLM) — например, GPT, Llama и пр. — отлично справляются с задачами на понимание и генерацию текста, я писал более детально про думающие модели типа o1 или DeepSeek R1 тут, однако:
Часто «галлюцинируют» (выдумывают детали),
Трудно оптимизируются для конкретного сложного пайплайна (например, когда задача требует нескольких шагов поиска, сводки, валидации).
Во многих случаях подход «один запрос – один ответ» оказывается недостаточным. Вместо этого на практике мы строим многоступенчатые пайплайны (compound AI systems), в которых каждый шаг решает конкретную подзадачу:
Сформировать поисковый запрос,
Найти или извлечь релевантные документы,
Свести результаты,
Сгенерировать финальный ответ.
Однако при создании подобной системы мы внезапно сталкиваемся с задачей:
Как формулировать промпты для каждого шага?
Как эффективно улучшать (оптимизировать) эти промпты, не имея индивидуальной метрики на каждом шаге (только итоговая метрика)?
Тут на помощь приходит фреймворк DSPy (Declarative Self-improving Python), предлагающий:
Описывать каждый этап в терминах «сигнатур» (signature) — что модуль принимает на вход, что должен вернуть, какое описание задачи,
Задавать «адаптер» (adapter), который превращает описание сигнатуры в реальный запрос к LLM,
Запускать «оптимизатор» (optimizer), который автоматически подбирает лучшие инструкции и лучшие few-shot-примеры для каждого модуля, чтобы максимизировать заданную итоговую метрику,
При необходимости использовать «бутстрап» (bootstrap) — извлечение демо-примеров из успешных прогона модели, и т.д.
Как обычно выглядит работа DSPy-оптимизаторов (Pipeline)
Если говорить именно о DSPy или схожих библиотеках, там процесс обычно строится так:
Adapter генерирует начальный промпт для каждого модуля (исходя из сигнатуры и предиктора). Это даёт базовую версию системы.
Мы генерируем примеры (примерно через rejection sampling) на тренировочном наборе: гоняем пайплайн и смотрим, где результат удовлетворяет метрике. Такие удачные траектории могут стать демо-примерами (бутстрап).
Оптимизатор начинает менять инструкции или набор демо-примеров (few-shot), используя:
Автоматический few-shot (dspy.BootstrapFewShotWithRandomSearch),
Индукцию инструкций (dspy.MIPROv2), OPRO, либо другие методы.
На каждом шаге оптимизатор пробует новые конфигурации, вызывает пайплайн на части тренировочного набора, измеряет метрику. По итогам он обновляет внутреннюю логику (например, в Bayesian TPE-стиле или через LLM, которые «учатся» на своих ошибках) и, наконец, выбирает лучшую конфигурацию промптов.
Таким образом, один и тот же модуль (скажем, «GenerateSearchQuery») может итеративно улучшаться:
Чётче формулировать инструкцию,
Добавлять лучшие примеры,
Исключать «вредные» или неработающие куски,
Согласовываться с другим модулем «AnswerWithContext».
Мейнстрим-подходы к оптимизации промптов
Существуют разные типы оптимизаторов, опирающиеся на идеи из статьи “Optimizing Instructions and Demonstrations for Multi-Stage Language Model Programs” (Khattab et al., 2024) и смежных работ:
Bootstrap Random Search
Сначала запускаем модель, собираем «удачные» примеры (input-output) и используем их в качестве кандидатов для few-shot.
Перебираем случайные наборы демо-примеров, выбирая те, что дают лучший score.
Module-Level OPRO (History-based)
Для каждого модуля ведём «историю» из [Instruction, Score].
Пробуем сгенерировать новые инструкции, ориентируясь на прошлые лучшие.
Небольшой минус: нет прямой оценки для каждого шага, поэтому результат может быть хуже, чем при согласованной оптимизации.
MIPRO / Bayesian Surrogate
Храним множество кандидатов (инструкций, демо-примеров) в некоем пуле.
Используем байесовский оптимизатор (TPE, Optuna), чтобы предсказывать: «если я выберу такие-то инструкции и демо-примеры, скорее всего итоговая метрика будет такой-то».
Выбираем наиболее перспективные комбинации, проверяем их реальным запуском.
По результатам обновляем нашу модель и движемся дальше. Обычно показывает себя мощно, так как ищет глобально лучшие комбинации.
OPRO (Program-level / Module-level)
Модель сама «учится» на истории результатов: видит список «[instr, score], [instr, score], ...» и пытается написать новую лучшую инструкцию.
Сложность: если пайплайн из нескольких модулей, нужно аккуратно решать задачу «credit assignment» (какой из шагов виноват в провале?).
Практическое задание
Задача: Реализовать (или продемонстрировать псевдокод) многошаговую систему «Вопрос–Ответ» (multi-hop QA) + применить автоматическую оптимизацию промптов для улучшения итоговой точности.
Система:
Модуль GenerateSearchQuery: принимает (question, context) и генерирует search_query.
Функция поиска: имитирует поиск, возвращает релевантные документы.
Модуль AnswerWithContext: собирает всё найденное и формирует финальный ответ.
Метрика:
Пусть это Exact Match с эталонным ответом (или любая своя).
Имеем тренировочный набор из (вопрос, правильный ответ).
Оптимизация:
Возьмём, к примеру, MIPRO или Bootstrap Random Search.
Соберём изначально демо-примеры (bootstrapping) или попробуем на голых инструкциях.
Итерируем, пока не найдём хороший набор инструкций + few-shot примеров.
Проверить на валидационном наборе, сравнить качество до и после.
Решение
Ниже приведён упрощённый пример. Он иллюстрирует, как мы можем описать модули (через сигнатуры) и как запустить DSPy-оптимизацию (например, с помощью MIPRO). Для корректного выполнения оптимизации нужно еще сконфигурировать языковую модель.
# https://dspy.ai/deep-dive/optimizers/miprov2/
import dspy
from dspy.teleprompt import MIPROv2
from dspy.datasets.gsm8k import GSM8K, gsm8k_metric
from dspy.evaluate import Evaluate
# ollama run
# deepseek-r1:14b
lm = dspy.LM('ollama_chat/falcon:7b',
api_base='http://localhost:11434', api_key='')
dspy.configure(lm=lm)
class CoT(dspy.Module):
def __init__(self):
super().__init__()
self.prog = dspy.ChainOfThought("question -> answer")
def forward(self, question):
return self.prog(question=question)
gms8k = GSM8K()
trainset, devset = gms8k.train, gms8k.dev
evaluate = Evaluate(devset=devset[:], metric=gsm8k_metric, num_threads=8, display_progress=True, display_table=False)
program = CoT()
evaluate(program, devset=devset[:])
teleprompter = MIPROv2(
metric=gsm8k_metric,
auto="light",
)
# Optimize program
print(f"Optimizing program with MIPRO...")
optimized_program = teleprompter.compile(
program.deepcopy(),
trainset=trainset,
max_bootstrapped_demos=3, # 0 for ZERO FEW-SHOT EXAMPLES
max_labeled_demos=4, # 0 ZERO FEW-SHOT EXAMPLES
requires_permission_to_run=False,
)
# Evaluate optimized program
print(f"Evaluate optimized program...")
evaluate(optimized_program, devset=devset[:])
optimized_program.save(f"mipro_optimized.json")
Этот код реализует модульную систему для ответа на вопросы:
Два шага:
Генерируется поисковой запрос на основе вопроса и текущего контекста.
После имитации поиска найденные данные добавляются к контексту, на основе которого формируется окончательный ответ.
Обучение:
Система обучается с использованием оптимизатора dspy.MIPROv2 и метрики точного совпадения на тренировочных данных.Тестирование:
После обучения система отвечает на тестовые вопросы, используя накопленный контекст.
Кому интересно совместно поработать над курсом по Интеллектуальным Агентам - пишите в личку.
