Цель — научиться создавать модульные (multi-stage) системы на базе LLM, а затем оптимизировать промпты (инструкции и примеры) таким образом, чтобы итоговая метрика качества (accuracy, retrieval score и т.п.) превышала вариант с ручным подбором текста промпта.
Почему нужны многошаговые LM-программы и оптимизация промпта
Современные большие языковые модели (LLM) — например, GPT, Llama и пр. — отлично справляются с задачами на понимание и генерацию текста, однако:
-
Часто «галлюцинируют» (выдумывают детали),
-
Трудно оптимизируются для конкретного сложного пайплайна (например, когда задача требует нескольких шагов поиска, сводки, валидации).
Во многих случаях подход «один запрос – один ответ» оказывается недостаточным. Вместо этого на практике мы строим многоступенчатые пайплайны (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). Для корректного выполнения оптимизации нужно еще сконфигурировать языковую модель.
import dspy
#############################################
# 1. Define Signatures for the Submodules #
#############################################
class GenerateSearchQuery(dspy.Signature):
context: str = dspy.InputField()
question: str = dspy.InputField()
search_query: str = dspy.OutputField()
class AnswerWithContext(dspy.Signature):
context: str = dspy.InputField()
question: str = dspy.InputField()
answer: str = dspy.OutputField()
#############################################
# 2. Define the MultiHopQA Module #
#############################################
class MultiHopQA(dspy.Module):
"""
Модульная система, состоящая из двух подпроцессов:
1) Генерация поискового запроса,
2) Генерация итогового ответа.
"""
def __init__(self, max_iterations=2):
super().__init__()
self.query_gen = dspy.ChainOfThought(GenerateSearchQuery)
self.answer_gen = dspy.ChainOfThought(AnswerWithContext)
self.max_iterations = max_iterations
def forward(self, question: str) -> str:
context_collected = ""
for _ in range(self.max_iterations):
# 1) Генерируем поисковый запрос
out_query = self.query_gen(
context=context_collected,
question=question
)
search_query = out_query["search_query"]
# 2) Имитируем поиск (или обращение к реальному поисковому движку)
retrieved_docs = f"Docs_for_{search_query}"
# 3) Добавляем найденное в общий контекст
context_collected += f"n{retrieved_docs}"
# 4) Генерируем финальный ответ
out_ans = self.answer_gen(
context=context_collected,
question=question
)
final_answer = out_ans["answer"]
return final_answer
#############################################
# 3. Prepare Data and Metric (Exact Match) #
#############################################
train_data = [
{
"input": "Кто написал роман 'Мастер и Маргарита'?",
"label": "Михаил Булгаков"
},
{
"input": "Какое уравнение открыл французский математик Жозеф Лиувилль?",
"label": "Уравнение Лиувилля (Liouville's equation)"
},
# ... можно добавить больше примеров
]
def exact_match(prediction: str, reference: str) -> float:
pred = prediction.strip().lower()
ref = reference.strip().lower()
return 1.0 if ref in pred else 0.0
def my_metric_fn(outputs, references) -> float:
scores = []
for pred, ref in zip(outputs, references):
scores.append(exact_match(pred, ref))
return sum(scores) / len(scores)
#############################################
# 4. Run the Optimizer (using MIPROv2) #
#############################################
pipeline = MultiHopQA(max_iterations=2)
# Функция для загрузки обучающих данных
def data_loader_fn():
inputs = [item["input"] for item in train_data]
references = [item["label"] for item in train_data]
return inputs, references
# Supply the required metric during instantiation.
optimizer = dspy.MIPROv2(metric=my_metric_fn)
# Use the compile method instead of optimize; pass the trainset directly.
trainset = data_loader_fn()
best_pipeline = optimizer.compile(pipeline, trainset=trainset, minibatch_size=1)
#############################################
# 5. Test the Final System #
#############################################
test_questions = [
"Кто написал роман 'Мастер и Маргарита'?",
"Каково основное уравнение Лиувилля?"
]
for q in test_questions:
ans = best_pipeline(q)
print(f"Вопрос: {q}nОтвет: {ans}n---")
Этот код реализует модульную систему для ответа на вопросы:
-
Два шага:
-
Генерируется поисковой запрос на основе вопроса и текущего контекста.
-
После имитации поиска найденные данные добавляются к контексту, на основе которого формируется окончательный ответ.
-
-
Обучение:
Система обучается с использованием оптимизатора dspy.MIPROv2 и метрики точного совпадения на тренировочных данных. -
Тестирование:
После обучения система отвечает на тестовые вопросы, используя накопленный контекст.
Кому интересно совместно поработать над курсом по Интеллектуальным Агентам - пишите в личку.
Автор: aufklarer