Dspy

DSPy: декларативные LM-программы, автооптимизация промптов, RAG.

Метаданные навыка

Источник	Опционально — установка: vibeos skills install official/mlops/dspy
Путь	optional-skills/mlops/research/dspy
Версия	1.0.0
Автор	Orchestra Research
Лицензия	MIT
Зависимости	dspy, openai, anthropic
Платформы	linux, macos, windows
Теги	Prompt Engineering, DSPy, Declarative Programming, RAG, Agents, Prompt Optimization, LM Programming, Stanford NLP, Automatic Optimization, Modular AI

Справочник: полный SKILL.md

к сведению

Ниже приведено полное определение навыка, которое VibeOS загружает при его активации. Это те инструкции, которые видит агент, когда навык активен.

DSPy: Декларативное программирование языковых моделей

Когда использовать этот навык

Используйте DSPy, когда вам нужно:

• Строить сложные AI-системы с несколькими компонентами и рабочими процессами
• Программировать LM декларативно вместо ручного инжиниринга промптов
• Автоматически оптимизировать промпты с помощью методов, основанных на данных
• Создавать модульные AI-пайплайны, которые поддерживаемы и переносимы
• Систематически улучшать выходные данные модели с помощью оптимизаторов
• Строить RAG-системы, агенты или классификаторы с повышенной надёжностью

Звёзды на GitHub: 22 000+ | Создатель: Stanford NLP

Установка

# Стабильный релиз
pip install dspy

# Последняя версия для разработки
pip install git+https://github.com/stanfordnlp/dspy.git

# С конкретными провайдерами LM
pip install dspy[openai]        # OpenAI
pip install dspy[anthropic]     # Anthropic Claude
pip install dspy[all]           # Все провайдеры

Быстрый старт

Базовый пример: ответы на вопросы

import dspy

# Настройка языковой модели
lm = dspy.Claude(model="claude-sonnet-4-5-20250929")
dspy.settings.configure(lm=lm)

# Определение сигнатуры (вход → выход)
class QA(dspy.Signature):
    """Отвечайте на вопросы короткими фактологическими ответами."""
    question = dspy.InputField()
    answer = dspy.OutputField(desc="обычно от 1 до 5 слов")

# Создание модуля
qa = dspy.Predict(QA)

# Использование
response = qa(question="Какая столица Франции?")
print(response.answer)  # "Париж"

Рассуждение по цепочке мыслей

import dspy

lm = dspy.Claude(model="claude-sonnet-4-5-20250929")
dspy.settings.configure(lm=lm)

# Использование ChainOfThought для лучшего рассуждения
class MathProblem(dspy.Signature):
    """Решайте текстовые математические задачи."""
    problem = dspy.InputField()
    answer = dspy.OutputField(desc="числовой ответ")

# ChainOfThought автоматически генерирует шаги рассуждения
cot = dspy.ChainOfThought(MathProblem)

response = cot(problem="У Джона 5 яблок, он отдаёт 2 Мэри. Сколько у него осталось?")
print(response.rationale)  # Показывает шаги рассуждения
print(response.answer)     # "3"

Основные концепции

1. Сигнатуры

Сигнатуры определяют структуру вашей AI-задачи (входы → выходы):

# Встроенная сигнатура (простая)
qa = dspy.Predict("question -> answer")

# Сигнатура-класс (подробная)
class Summarize(dspy.Signature):
    """Суммируйте текст в ключевые пункты."""
    text = dspy.InputField()
    summary = dspy.OutputField(desc="маркированный список, 3–5 пунктов")

summarizer = dspy.ChainOfThought(Summarize)

Когда что использовать:

• Встроенная: Быстрое прототипирование, простые задачи
• Класс: Сложные задачи, подсказки типов, лучшая документация

2. Модули

Модули — это переиспользуемые компоненты, преобразующие входы в выходы:

dspy.Predict

Базовый модуль предсказания:

predictor = dspy.Predict("context, question -> answer")
result = predictor(context="Париж — столица Франции",
                   question="Какая столица?")

dspy.ChainOfThought

Генерирует шаги рассуждения перед ответом:

cot = dspy.ChainOfThought("question -> answer")
result = cot(question="Почему небо голубое?")
print(result.rationale)  # Шаги рассуждения
print(result.answer)     # Финальный ответ

dspy.ReAct

Агентоподобное рассуждение с инструментами:

from dspy.predict import ReAct

class SearchQA(dspy.Signature):
    """Отвечайте на вопросы, используя поиск."""
    question = dspy.InputField()
    answer = dspy.OutputField()

def search_tool(query: str) -> str:
    """Поиск в Википедии."""
    # Ваша реализация поиска
    return results

react = ReAct(SearchQA, tools=[search_tool])
result = react(question="Когда был создан Python?")

dspy.ProgramOfThought

Генерирует и выполняет код для рассуждения:

pot = dspy.ProgramOfThought("question -> answer")
result = pot(question="Сколько будет 15% от 240?")
# Генерирует: answer = 240 * 0.15

3. Оптимизаторы

Оптимизаторы автоматически улучшают ваши модули, используя обучающие данные:

BootstrapFewShot

Учится на примерах:

from dspy.teleprompt import BootstrapFewShot

# Обучающие данные
trainset = [
    dspy.Example(question="Сколько будет 2+2?", answer="4").with_inputs("question"),
    dspy.Example(question="Сколько будет 3+5?", answer="8").with_inputs("question"),
]

# Определение метрики
def validate_answer(example, pred, trace=None):
    return example.answer == pred.answer

# Оптимизация
optimizer = BootstrapFewShot(metric=validate_answer, max_bootstrapped_demos=3)
optimized_qa = optimizer.compile(qa, trainset=trainset)

# Теперь optimized_qa работает лучше!

MIPRO (Most Important Prompt Optimization)

Итеративно улучшает промпты:

from dspy.teleprompt import MIPRO

optimizer = MIPRO(
    metric=validate_answer,
    num_candidates=10,
    init_temperature=1.0
)

optimized_cot = optimizer.compile(
    cot,
    trainset=trainset,
    num_trials=100
)

BootstrapFinetune

Создаёт наборы данных для тонкой настройки модели:

from dspy.teleprompt import BootstrapFinetune

optimizer = BootstrapFinetune(metric=validate_answer)
optimized_module = optimizer.compile(qa, trainset=trainset)

# Экспортирует обучающие данные для тонкой настройки

4. Построение сложных систем

Многоэтапный пайплайн

import dspy

class MultiHopQA(dspy.Module):
    def __init__(self):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=3)
        self.generate_query = dspy.ChainOfThought("question -> search_query")
        self.generate_answer = dspy.ChainOfThought("context, question -> answer")

    def forward(self, question):
        # Этап 1: Генерация поискового запроса
        search_query = self.generate_query(question=question).search_query

        # Этап 2: Извлечение контекста
        passages = self.retrieve(search_query).passages
        context = "\n".join(passages)

        # Этап 3: Генерация ответа
        answer = self.generate_answer(context=context, question=question).answer
        return dspy.Prediction(answer=answer, context=context)

# Использование пайплайна
qa_system = MultiHopQA()
result = qa_system(question="Кто написал книгу, вдохновившую фильм «Бегущий по лезвию»?")

RAG-система с оптимизацией

import dspy
from dspy.retrieve.chromadb_rm import ChromadbRM

# Настройка ретривера
retriever = ChromadbRM(
    collection_name="documents",
    persist_directory="./chroma_db"
)

class RAG(dspy.Module):
    def __init__(self, num_passages=3):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=num_passages)
        self.generate = dspy.ChainOfThought("context, question -> answer")

    def forward(self, question):
        context = self.retrieve(question).passages
        return self.generate(context=context, question=question)

# Создание и оптимизация
rag = RAG()

# Оптимизация с обучающими данными
from dspy.teleprompt import BootstrapFewShot

optimizer = BootstrapFewShot(metric=validate_answer)
optimized_rag = optimizer.compile(rag, trainset=trainset)

Конфигурация провайдеров LM

Anthropic Claude

import dspy

lm = dspy.Claude(
    model="claude-sonnet-4-5-20250929",
    api_key="ваш-api-ключ",  # Или установите переменную окружения ANTHROPIC_API_KEY
    max_tokens=1000,
    temperature=0.7
)
dspy.settings.configure(lm=lm)

OpenAI

lm = dspy.OpenAI(
    model="gpt-4",
    api_key="ваш-api-ключ",
    max_tokens=1000
)
dspy.settings.configure(lm=lm)

Локальные модели (Ollama)

lm = dspy.OllamaLocal(
    model="llama3.1",
    base_url="http://localhost:11434"
)
dspy.settings.configure(lm=lm)

Несколько моделей

# Разные модели для разных задач
cheap_lm = dspy.OpenAI(model="gpt-3.5-turbo")
strong_lm = dspy.Claude(model="claude-sonnet-4-5-20250929")

# Использование дешёвой модели для извлечения, сильной — для рассуждения
with dspy.settings.context(lm=cheap_lm):
    context = retriever(question)

with dspy.settings.context(lm=strong_lm):
    answer = generator(context=context, question=question)

Типовые паттерны

Паттерн 1: Структурированный вывод

from pydantic import BaseModel, Field

class PersonInfo(BaseModel):
    name: str = Field(description="Полное имя")
    age: int = Field(description="Возраст в годах")
    occupation: str = Field(description="Текущая работа")

class ExtractPerson(dspy.Signature):
    """Извлеките информацию о человеке из текста."""
    text = dspy.InputField()
    person: PersonInfo = dspy.OutputField()

extractor = dspy.TypedPredictor(ExtractPerson)
result = extractor(text="Иван Петров — 35-летний инженер-программист.")
print(result.person.name)  # "Иван Петров"
print(result.person.age)   # 35

Паттерн 2: Оптимизация на основе утверждений

import dspy
from dspy.primitives.assertions import assert_transform_module, backtrack_handler

class MathQA(dspy.Module):
    def __init__(self):
        super().__init__()
        self.solve = dspy.ChainOfThought("problem -> solution: float")

    def forward(self, problem):
        solution = self.solve(problem=problem).solution

        # Утверждение, что решение является числовым
        dspy.Assert(
            isinstance(float(solution), float),
            "Решение должно быть числом",
            backtrack=backtrack_handler
        )

        return dspy.Prediction(solution=solution)

Паттерн 3: Самосогласованность

import dspy
from collections import Counter

class ConsistentQA(dspy.Module):
    def __init__(self, num_samples=5):
        super().__init__()
        self.qa = dspy.ChainOfThought("question -> answer")
        self.num_samples = num_samples

    def forward(self, question):
        # Генерация нескольких ответов
        answers = []
        for _ in range(self.num_samples):
            result = self.qa(question=question)
            answers.append(result.answer)

        # Возврат самого частого ответа
        most_common = Counter(answers).most_common(1)[0][0]
        return dspy.Prediction(answer=most_common)

Паттерн 4: Извлечение с переранжированием

class RerankedRAG(dspy.Module):
    def __init__(self):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=10)
        self.rerank = dspy.Predict("question, passage -> relevance_score: float")
        self.answer = dspy.ChainOfThought("context, question -> answer")

    def forward(self, question):
        # Извлечение кандидатов
        passages = self.retrieve(question).passages

        # Переранжирование отрывков
        scored = []
        for passage in passages:
            score = float(self.rerank(question=question, passage=passage).relevance_score)
            scored.append((score, passage))

        # Взять топ-3
        top_passages = [p for _, p in sorted(scored, reverse=True)[:3]]
        context = "\n\n".join(top_passages)

        # Генерация ответа
        return self.answer(context=context, question=question)

Оценка и метрики

Пользовательские метрики

def exact_match(example, pred, trace=None):
    """Метрика точного совпадения."""
    return example.answer.lower() == pred.answer.lower()

def f1_score(example, pred, trace=None):
    """F1-мера для перекрытия текста."""
    pred_tokens = set(pred.answer.lower().split())
    gold_tokens = set(example.answer.lower().split())

    if not pred_tokens:
        return 0.0

    precision = len(pred_tokens & gold_tokens) / len(pred_tokens)
    recall = len(pred_tokens & gold_tokens) / len(gold_tokens)

    if precision + recall == 0:
        return 0.0

    return 2 * (precision * recall) / (precision + recall)

Оценка

from dspy.evaluate import Evaluate

# Создание оценщика
evaluator = Evaluate(
    devset=testset,
    metric=exact_match,
    num_threads=4,
    display_progress=True
)

# Оценка модели
score = evaluator(qa_system)
print(f"Точность: {score}")

# Сравнение оптимизированной и неоптимизированной версий
score_before = evaluator(qa)
score_after = evaluator(optimized_qa)
print(f"Улучшение: {score_after - score_before:.2%}")

Лучшие практики

1. Начинайте с простого, итерируйте

# Начните с Predict
qa = dspy.Predict("question -> answer")

# Добавьте рассуждение, если нужно
qa = dspy.ChainOfThought("question -> answer")

# Добавьте оптимизацию, когда есть данные
optimized_qa = optimizer.compile(qa, trainset=data)

2. Используйте описательные сигнатуры

# ❌ Плохо: расплывчато
class Task(dspy.Signature):
    input = dspy.InputField()
    output = dspy.OutputField()

# ✅ Хорошо: описательно
class SummarizeArticle(dspy.Signature):
    """Суммируйте новостные статьи в 3–5 ключевых пунктов."""
    article = dspy.InputField(desc="полный текст статьи")
    summary = dspy.OutputField(desc="маркированный список, 3–5 пунктов")

3. Оптимизируйте на репрезентативных данных

# Создайте разнообразные обучающие примеры
trainset = [
    dspy.Example(question="фактологический", answer="...").with_inputs("question"),
    dspy.Example(question="рассуждение", answer="...").with_inputs("question"),
    dspy.Example(question="вычисление", answer="...").with_inputs("question"),
]

# Используйте валидационный набор для метрики
def metric(example, pred, trace=None):
    return example.answer in pred.answer

4. Сохраняйте и загружайте оптимизированные модели

# Сохранение
optimized_qa.save("models/qa_v1.json")

# Загрузка
loaded_qa = dspy.ChainOfThought("question -> answer")
loaded_qa.load("models/qa_v1.json")

5. Мониторинг и отладка

# Включение трассировки
dspy.settings.configure(lm=lm, trace=[])

# Запуск предсказания
result = qa(question="...")

# Просмотр трассировки
for call in dspy.settings.trace:
    print(f"Промпт: {call['prompt']}")
    print(f"Ответ: {call['response']}")

Сравнение с другими подходами

Особенность	Ручное написание промптов	LangChain	DSPy
Инжиниринг промптов	Вручную	Вручную	Автоматически
Оптимизация	Методом проб и ошибок	Нет	На основе данных
Модульность	Низкая	Средняя	Высокая
Безопасность типов	Нет	Ограниченная	Да (сигнатуры)
Переносимость	Низкая	Средняя	Высокая
Кривая обучения	Низкая	Средняя	Средняя–высокая

Когда выбирать DSPy:

• У вас есть обучающие данные или возможность их сгенерировать
• Вам нужно систематическое улучшение промптов
• Вы строите сложные многоэтапные системы
• Вы хотите оптимизировать для разных LM

Когда выбирать альтернативы:

• Быстрые прототипы (ручное написание промптов)
• Простые цепочки с существующими инструментами (LangChain)
• Требуется собственная логика оптимизации

Ресурсы

• Документация: https://dspy.ai
• GitHub: https://github.com/stanfordnlp/dspy (22k+ звёзд)
• Discord: https://discord.gg/XCGy2WDCQB
• Twitter: @DSPyOSS
• Статья: «DSPy: Compiling Declarative Language Model Calls into Self-Improving Pipelines»

См. также

• references/modules.md — Подробное руководство по модулям (Predict, ChainOfThought, ReAct, ProgramOfThought)
• references/optimizers.md — Алгоритмы оптимизации (BootstrapFewShot, MIPRO, BootstrapFinetune)
• references/examples.md — Примеры из реального мира (RAG, агенты, классификаторы)