Определение сигнатур и модулей

Сигнатуры — типизированные контракты

Определение базовой сигнатуры

import dspy

class QASignature(dspy.Signature):
    """Answer the question based on the given context."""
    context: str = dspy.InputField(desc='Relevant background text')
    question: str = dspy.InputField(desc='The question to answer')
    answer: str = dspy.OutputField(desc='A concise answer')

# Inspect what DSPy sees
print(QASignature.instructions)  # The docstring
print(list(QASignature.input_fields.keys()))   # ['context', 'question']
print(list(QASignature.output_fields.keys()))  # ['answer']

Несколько выходных полей

import dspy

class EntityExtraction(dspy.Signature):
    """Extract named entities from the text."""
    text: str = dspy.InputField()
    people: list[str] = dspy.OutputField(desc='List of person names mentioned')
    organizations: list[str] = dspy.OutputField(desc='List of organization names')
    locations: list[str] = dspy.OutputField(desc='List of place names')

extractor = dspy.Predict(EntityExtraction)
result = extractor(text='Elon Musk founded SpaceX in Hawthorne, California.')
print(result.people, result.organizations, result.locations)

Модуль Predict

import dspy

class Classify(dspy.Signature):
    """Classify the sentiment of the review."""
    review: str = dspy.InputField()
    sentiment: str = dspy.OutputField(desc='positive, negative, or neutral')

# Predict wraps the signature with a direct prompt
classifier = dspy.Predict(Classify)
result = classifier(review='The food was amazing and the service was excellent!')
print(result.sentiment)  # positive

Модуль ChainOfThought

import dspy

class MathSolver(dspy.Signature):
    """Solve the math problem."""
    problem: str = dspy.InputField()
    answer: str = dspy.OutputField(desc='The numerical answer')

# ChainOfThought adds a 'reasoning' step automatically
solver = dspy.ChainOfThought(MathSolver)
result = solver(problem='If a train travels 60 mph for 2.5 hours, how far does it go?')
print(result.reasoning)  # Step-by-step reasoning
print(result.answer)     # 150 miles

Модуль ReAct

import dspy

def search_web(query: str) -> str:
    """Search the web and return relevant results."""
    # In production, call a real search API
    return f'Search results for: {query}'

def calculate(expression: str) -> str:
    """Evaluate a mathematical expression."""
    return str(eval(expression))

class ResearchQA(dspy.Signature):
    """Answer the question using web search and calculation as needed."""
    question: str = dspy.InputField()
    answer: str = dspy.OutputField()

agent = dspy.ReAct(ResearchQA, tools=[search_web, calculate])
result = agent(question='What is 15% of 847?')
print(result.answer)

Объединение модулей в программы

import dspy

class RetrieveAndAnswer(dspy.Module):
    def __init__(self):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=3)  # Retrieves top-3 passages
        self.generate = dspy.ChainOfThought('context, question -> answer')

    def forward(self, question):
        passages = self.retrieve(question).passages
        context = '\n'.join(passages)
        return self.generate(context=context, question=question)

# This is a complete RAG pipeline in ~10 lines
rag = RetrieveAndAnswer()
result = rag(question='What are the main causes of climate change?')
print(result.answer)

Сокращённая запись сигнатуры в строке

import dspy

# Full class signature
class Translate(dspy.Signature):
    """Translate text to French."""
    text: str = dspy.InputField()
    translation: str = dspy.OutputField()

# Equivalent inline shorthand
translator_v1 = dspy.Predict(Translate)
translator_v2 = dspy.Predict('text -> translation')  # Less metadata

# Both work the same way
result = translator_v1(text='Hello world')
print(result.translation)

Описания полей важны

import dspy

class Summarize(dspy.Signature):
    """Summarize the article for a busy executive."""
    article: str = dspy.InputField(
        desc='The full article text to summarize'
    )
    summary: str = dspy.OutputField(
        desc='3-5 bullet points highlighting key decisions and numbers'
    )
    confidence: float = dspy.OutputField(
        desc='Your confidence in the summary accuracy from 0.0 to 1.0'
    )

summarizer = dspy.Predict(Summarize)
# DSPy constructs a prompt using all the desc values automatically

Сохранение и загрузка программ

import dspy

class QA(dspy.Signature):
    """Answer questions accurately."""
    question: str = dspy.InputField()
    answer: str = dspy.OutputField()

program = dspy.ChainOfThought(QA)

# After optimization, save the compiled state
program.save('optimized_qa.json')

# Load it later without re-running optimization
loaded_program = dspy.ChainOfThought(QA)
loaded_program.load('optimized_qa.json')

result = loaded_program(question='What is the speed of light?')
print(result.answer)

Утверждения для ограничения выходных данных

import dspy

class ShortAnswer(dspy.Signature):
    """Answer in at most 10 words."""
    question: str = dspy.InputField()
    answer: str = dspy.OutputField()

class ConstrainedQA(dspy.Module):
    def __init__(self):
        super().__init__()
        self.predict = dspy.Predict(ShortAnswer)

    def forward(self, question):
        result = self.predict(question=question)
        # Assert: answer must be at most 10 words
        dspy.Assert(
            len(result.answer.split()) <= 10,
            'The answer must be 10 words or fewer.'
        )
        return result

Проверка знаний: ChainOfThought

Повторение: сигнатуры и модули