Chapter 09: DSPy Advanced Patterns and Techniques
Haiyue
38min
Chapter 09: DSPy Advanced Patterns and Techniques
Learning Objectives
- Implement program synthesis and automatic programming
- Learn meta-learning and few-shot learning techniques
- Master advanced strategies for prompt optimization
- Explore self-consistency and voting mechanisms
- Implement dynamic prompt generation
Key Concepts
1. Program Synthesis and Automatic Programming
DSPy can implement automatic program generation and synthesis, which is cutting-edge AI programming technology.
Basic Framework for Program Synthesis
import dspy
from typing import List, Dict, Any, Optional, Union, Callable
import ast
import inspect
import json
import re
from dataclasses import dataclass
from abc import ABC, abstractmethod
class ProgramComponent(ABC):
"""Base class for program components"""
def __init__(self, name: str, description: str = ""):
self.name = name
self.description = description
self.inputs = []
self.outputs = []
self.dependencies = []
@abstractmethod
def generate_code(self, context: Dict[str, Any]) -> str:
"""Generate code"""
pass
@abstractmethod
def get_signature(self) -> str:
"""Get function signature"""
pass
class DSPyComponent(ProgramComponent):
"""DSPy component"""
def __init__(self, name: str, signature: str, instructions: str = ""):
super().__init__(name, f"DSPy module: {signature}")
self.signature = signature
self.instructions = instructions
self.module_type = "ChainOfThought" # Default type
def generate_code(self, context: Dict[str, Any]) -> str:
"""Generate DSPy module code"""
code_template = f"""
# {self.description}
self.{self.name} = dspy.{self.module_type}(
"{self.signature}",
instructions=\"\"\"{self.instructions}\"\"\"
)
"""
return code_template.strip()
def get_signature(self) -> str:
return self.signature
class ProgramSynthesizer(dspy.Module):
"""Program synthesizer"""
def __init__(self):
super().__init__()
# Requirement analysis module
self.requirement_analyzer = dspy.ChainOfThought(
"task_description -> reasoning, requirements",
instructions="Analyze task description to identify specific functional requirements and constraints."
)
# Architecture design module
self.architecture_designer = dspy.ChainOfThought(
"requirements -> reasoning, architecture",
instructions="Design program architecture based on requirements, determining needed components and their relationships."
)
# Code generation module
self.code_generator = dspy.ChainOfThought(
"architecture, component_spec -> reasoning, code",
instructions="Generate Python code based on architecture and component specifications."
)
# Code optimization module
self.code_optimizer = dspy.ChainOfThought(
"raw_code, optimization_goals -> reasoning, optimized_code",
instructions="Optimize generated code to improve efficiency and readability."
)
# Component library
self.component_library = {
'text_processor': DSPyComponent(
"text_processor",
"text -> processed_text",
"Process and clean input text"
),
'question_answerer': DSPyComponent(
"question_answerer",
"question, context -> answer",
"Answer questions based on context"
),
'summarizer': DSPyComponent(
"summarizer",
"long_text -> summary",
"Generate text summary"
),
'classifier': DSPyComponent(
"classifier",
"text -> category",
"Classify text"
)
}
def forward(self, task_description: str):
"""Automatic program synthesis"""
print(f"🎯 Starting program synthesis: {task_description}")
# 1. Requirement analysis
requirements_result = self.requirement_analyzer(
task_description=task_description
)
print(f"📋 Requirement analysis: {requirements_result.requirements}")
# 2. Architecture design
architecture_result = self.architecture_designer(
requirements=requirements_result.requirements
)
print(f"🏗️ Architecture design: {architecture_result.architecture}")
# 3. Select and combine components
selected_components = self.select_components(
requirements_result.requirements,
architecture_result.architecture
)
print(f"🧩 Selected components: {[comp.name for comp in selected_components]}")
# 4. Generate code
synthesized_program = self.synthesize_program(
task_description,
selected_components,
architecture_result.architecture
)
# 5. Optimize code
optimized_program = self.optimize_program(synthesized_program)
return dspy.Prediction(
task_description=task_description,
requirements=requirements_result.requirements,
architecture=architecture_result.architecture,
selected_components=[comp.name for comp in selected_components],
synthesized_program=synthesized_program,
optimized_program=optimized_program
)
def select_components(self,
requirements: str,
architecture: str) -> List[ProgramComponent]:
"""Select appropriate components"""
# Component selection based on keyword matching
requirement_text = (requirements + " " + architecture).lower()
selected_components = []
# Check if various components are needed
if any(keyword in requirement_text for keyword in ['qa', 'question', 'answer']):
selected_components.append(self.component_library['question_answerer'])
if any(keyword in requirement_text for keyword in ['summary', 'summarize']):
selected_components.append(self.component_library['summarizer'])
if any(keyword in requirement_text for keyword in ['classify', 'category', 'classification']):
selected_components.append(self.component_library['classifier'])
if any(keyword in requirement_text for keyword in ['process', 'clean']):
selected_components.append(self.component_library['text_processor'])
# If no components selected, add text processor by default
if not selected_components:
selected_components.append(self.component_library['text_processor'])
return selected_components
def synthesize_program(self,
task_description: str,
components: List[ProgramComponent],
architecture: str) -> str:
"""Synthesize program"""
# Generate class definition
class_name = "SynthesizedProgram"
program_parts = [
f"import dspy\n",
f"class {class_name}(dspy.Module):",
f' """Automatically synthesized program: {task_description}"""',
f" ",
f" def __init__(self):",
f" super().__init__()",
f" "
]
# Generate component initialization code
for component in components:
component_code = component.generate_code({})
program_parts.extend([
f" {component_code}",
f" "
])
# Generate forward method
program_parts.extend([
f" def forward(self, **kwargs):",
f" # Automatically generated execution logic",
f" "
])
# Generate execution logic
forward_logic = self.generate_forward_logic(components, architecture)
program_parts.extend([
f" {forward_logic}",
f" ",
f" return dspy.Prediction(**results)"
])
return "\n".join(program_parts)
def generate_forward_logic(self,
components: List[ProgramComponent],
architecture: str) -> str:
"""Generate forward method logic"""
if len(components) == 1:
comp = components[0]
return f"""
result = self.{comp.name}(**kwargs)
results = {{'output': result}}
""".strip()
# Sequential execution logic for multiple components
logic_parts = [
"results = {}",
"intermediate_results = {}"
]
for i, comp in enumerate(components):
if i == 0:
logic_parts.append(f"result_{i} = self.{comp.name}(**kwargs)")
else:
logic_parts.append(f"result_{i} = self.{comp.name}(**intermediate_results)")
logic_parts.append(f"intermediate_results.update(result_{i}.__dict__)")
logic_parts.append(f"results['final_output'] = result_{len(components)-1}")
return "\n ".join(logic_parts)
def optimize_program(self, raw_program: str) -> str:
"""Optimize generated program"""
optimization_goals = "Improve code readability, optimize performance, add error handling"
optimization_result = self.code_optimizer(
raw_code=raw_program,
optimization_goals=optimization_goals
)
return optimization_result.optimized_code
class MetaProgramming:
"""Meta-programming tools"""
def __init__(self):
self.program_templates = {}
self.generated_programs = {}
def register_template(self,
template_name: str,
template_code: str,
parameters: List[str]):
"""Register program template"""
self.program_templates[template_name] = {
'code': template_code,
'parameters': parameters,
'usage_count': 0
}
print(f"📝 Registered program template: {template_name}")
def instantiate_template(self,
template_name: str,
parameter_values: Dict[str, Any]) -> str:
"""Instantiate template"""
if template_name not in self.program_templates:
raise ValueError(f"Template {template_name} does not exist")
template = self.program_templates[template_name]
template_code = template['code']
# Parameter replacement
for param, value in parameter_values.items():
placeholder = f"{{{{{param}}}}}"
template_code = template_code.replace(placeholder, str(value))
# Update usage count
template['usage_count'] += 1
# Generate unique program ID
program_id = f"{template_name}_{template['usage_count']}"
self.generated_programs[program_id] = template_code
print(f"🔧 Instantiated template: {template_name} -> {program_id}")
return template_code
def create_program_from_examples(self,
examples: List[dspy.Example],
program_type: str = "classifier") -> str:
"""Create program from examples"""
example_analyzer = dspy.ChainOfThought(
"examples -> reasoning, program_structure",
instructions="Analyze given examples to infer the program structure and logic."
)
examples_text = self.format_examples_for_analysis(examples)
analysis_result = example_analyzer(examples=examples_text)
# Generate program based on analysis results
if program_type == "classifier":
return self.generate_classifier_program(analysis_result.program_structure)
elif program_type == "qa":
return self.generate_qa_program(analysis_result.program_structure)
else:
return self.generate_generic_program(analysis_result.program_structure)
def format_examples_for_analysis(self, examples: List[dspy.Example]) -> str:
"""Format examples for analysis"""
formatted_examples = []
for i, example in enumerate(examples, 1):
inputs = example.inputs()
outputs = {k: v for k, v in example.__dict__.items() if k not in inputs}
formatted_examples.append(f"Example {i}:")
formatted_examples.append(f" Input: {inputs}")
formatted_examples.append(f" Output: {outputs}")
return "\n".join(formatted_examples)
def generate_classifier_program(self, structure_description: str) -> str:
"""Generate classifier program"""
template = """
import dspy
class AutoGeneratedClassifier(dspy.Module):
def __init__(self):
super().__init__()
self.classifier = dspy.ChainOfThought(
"{{input_fields}} -> reasoning, category",
instructions="{{classification_instructions}}"
)
def forward(self, **kwargs):
result = self.classifier(**kwargs)
return dspy.Prediction(category=result.category, reasoning=result.reasoning)
"""
# Further customize template based on structure_description
return template
def generate_qa_program(self, structure_description: str) -> str:
"""Generate Q&A program"""
template = """
import dspy
class AutoGeneratedQA(dspy.Module):
def __init__(self):
super().__init__()
self.qa = dspy.ChainOfThought(
"question, context -> reasoning, answer",
instructions="Answer questions based on given context."
)
def forward(self, question, context=None, **kwargs):
result = self.qa(question=question, context=context or "")
return dspy.Prediction(answer=result.answer, reasoning=result.reasoning)
"""
return template
def generate_generic_program(self, structure_description: str) -> str:
"""Generate generic program"""
template = """
import dspy
class AutoGeneratedProgram(dspy.Module):
def __init__(self):
super().__init__()
self.processor = dspy.ChainOfThought(
"input -> reasoning, output",
instructions="Process input and generate corresponding output."
)
def forward(self, **kwargs):
result = self.processor(**kwargs)
return result
"""
return template
# Usage examples
def demonstrate_program_synthesis():
"""Demonstrate program synthesis"""
synthesizer = ProgramSynthesizer()
# Test task descriptions
task_descriptions = [
"Create a text classification system that can categorize news articles into sports, technology, politics, etc.",
"Build a Q&A system that can answer user questions based on given documents",
"Develop a text summarization tool that can compress long texts into brief summaries"
]
for task_desc in task_descriptions:
print(f"\n{'='*60}")
result = synthesizer(task_desc)
print(f"📊 Synthesis result:")
print(f"Requirements: {result.requirements}")
print(f"Architecture: {result.architecture}")
print(f"Components: {result.selected_components}")
print(f"\n🔍 Generated program:")
print(result.optimized_program[:500] + "..." if len(result.optimized_program) > 500 else result.optimized_program)
return synthesizer
# demo_synthesis = demonstrate_program_synthesis()2. Meta-Learning and Few-Shot Learning
Meta-learning enables models to quickly adapt to new tasks, and few-shot learning achieves good results with minimal examples.
class MetaLearner(dspy.Module):
"""Meta-learner"""
def __init__(self):
super().__init__()
# Task understanding module
self.task_analyzer = dspy.ChainOfThought(
"task_examples -> reasoning, task_pattern",
instructions="Analyze task examples to identify task patterns and regularities."
)
# Strategy generation module
self.strategy_generator = dspy.ChainOfThought(
"task_pattern, available_examples -> reasoning, learning_strategy",
instructions="Based on task patterns and available examples, formulate optimal learning strategy."
)
# Fast adaptation module
self.fast_adaptor = dspy.ChainOfThought(
"new_example, learned_pattern, strategy -> reasoning, adapted_response",
instructions="Based on learned patterns and strategies, quickly adapt to handle new examples."
)
# Meta-knowledge base
self.meta_knowledge = {
'task_patterns': {},
'successful_strategies': {},
'adaptation_history': []
}
def learn_from_tasks(self, tasks: List[Dict[str, Any]]) -> Dict[str, Any]:
"""Learn from multiple tasks"""
print(f"🎓 Starting meta-learning, number of tasks: {len(tasks)}")
learned_patterns = {}
successful_strategies = {}
for i, task in enumerate(tasks):
print(f"\n📚 Learning task {i+1}: {task.get('name', f'Task_{i+1}')}")
# Analyze task pattern
examples_text = self.format_task_examples(task['examples'])
pattern_result = self.task_analyzer(task_examples=examples_text)
# Generate learning strategy
strategy_result = self.strategy_generator(
task_pattern=pattern_result.task_pattern,
available_examples=examples_text
)
# Store learning results
task_type = task.get('type', f'task_{i}')
learned_patterns[task_type] = pattern_result.task_pattern
successful_strategies[task_type] = strategy_result.learning_strategy
print(f"🔍 Task pattern: {pattern_result.task_pattern[:100]}...")
print(f"📋 Learning strategy: {strategy_result.learning_strategy[:100]}...")
# Update meta-knowledge base
self.meta_knowledge['task_patterns'].update(learned_patterns)
self.meta_knowledge['successful_strategies'].update(successful_strategies)
# Extract common patterns
common_patterns = self.extract_common_patterns(learned_patterns)
return {
'learned_patterns': learned_patterns,
'successful_strategies': successful_strategies,
'common_patterns': common_patterns,
'meta_insights': self.generate_meta_insights(learned_patterns, successful_strategies)
}
def format_task_examples(self, examples: List[dspy.Example]) -> str:
"""Format task examples"""
formatted_examples = []
for i, example in enumerate(examples[:5]): # Limit number of examples
inputs = example.inputs()
outputs = {k: v for k, v in example.__dict__.items() if k not in inputs}
formatted_examples.append(f"Example {i+1}:")
formatted_examples.append(f" Input: {inputs}")
formatted_examples.append(f" Output: {outputs}")
return "\n".join(formatted_examples)
def extract_common_patterns(self, patterns: Dict[str, str]) -> List[str]:
"""Extract common patterns"""
# Simplified pattern extraction (actual implementation may need more complex NLP analysis)
all_patterns_text = " ".join(patterns.values()).lower()
# Find frequently occurring keywords and phrases
words = all_patterns_text.split()
word_freq = {}
for word in words:
if len(word) > 3: # Ignore short words
word_freq[word] = word_freq.get(word, 0) + 1
# Extract high-frequency words as common patterns
common_patterns = [
word for word, freq in word_freq.items()
if freq >= len(patterns) * 0.5 # Appears in at least half of patterns
]
return common_patterns[:10] # Return top 10 common patterns
def generate_meta_insights(self,
patterns: Dict[str, str],
strategies: Dict[str, str]) -> str:
"""Generate meta-insights"""
meta_insight_generator = dspy.ChainOfThought(
"patterns, strategies -> reasoning, meta_insights",
instructions="Based on learned patterns and strategies, generate meta-insights about task types and learning methods."
)
result = meta_insight_generator(
patterns=json.dumps(patterns, ensure_ascii=False),
strategies=json.dumps(strategies, ensure_ascii=False)
)
return result.meta_insights
def adapt_to_new_task(self,
new_examples: List[dspy.Example],
task_type_hint: str = None) -> Dict[str, Any]:
"""Adapt to new task"""
print(f"🎯 Quickly adapting to new task")
# Analyze new task
new_examples_text = self.format_task_examples(new_examples)
new_pattern_result = self.task_analyzer(task_examples=new_examples_text)
# Find similar known pattern
similar_pattern = self.find_similar_pattern(
new_pattern_result.task_pattern,
task_type_hint
)
if similar_pattern:
print(f"🔍 Found similar pattern: {similar_pattern['type']}")
# Use strategy from similar pattern
adaptation_strategy = similar_pattern['strategy']
else:
print(f"🆕 This is a completely new task pattern")
# Generate strategy for new pattern
strategy_result = self.strategy_generator(
task_pattern=new_pattern_result.task_pattern,
available_examples=new_examples_text
)
adaptation_strategy = strategy_result.learning_strategy
# Create adapted processor
adapted_processor = self.create_adapted_processor(
new_pattern_result.task_pattern,
adaptation_strategy,
new_examples
)
return {
'task_pattern': new_pattern_result.task_pattern,
'adaptation_strategy': adaptation_strategy,
'adapted_processor': adapted_processor,
'confidence': self.calculate_adaptation_confidence(similar_pattern)
}
def find_similar_pattern(self,
new_pattern: str,
hint: str = None) -> Optional[Dict[str, Any]]:
"""Find similar pattern"""
if not self.meta_knowledge['task_patterns']:
return None
# If hint provided, prioritize finding corresponding pattern
if hint and hint in self.meta_knowledge['task_patterns']:
return {
'type': hint,
'pattern': self.meta_knowledge['task_patterns'][hint],
'strategy': self.meta_knowledge['successful_strategies'].get(hint, "")
}
# Simplified similarity calculation (based on keyword overlap)
best_similarity = 0.0
most_similar = None
new_words = set(new_pattern.lower().split())
for task_type, pattern in self.meta_knowledge['task_patterns'].items():
pattern_words = set(pattern.lower().split())
if new_words and pattern_words:
overlap = len(new_words & pattern_words)
union = len(new_words | pattern_words)
similarity = overlap / union if union > 0 else 0.0
if similarity > best_similarity and similarity > 0.3: # Threshold
best_similarity = similarity
most_similar = {
'type': task_type,
'pattern': pattern,
'strategy': self.meta_knowledge['successful_strategies'].get(task_type, ""),
'similarity': similarity
}
return most_similar
def create_adapted_processor(self,
pattern: str,
strategy: str,
examples: List[dspy.Example]) -> dspy.Module:
"""Create adapted processor"""
class AdaptedProcessor(dspy.Module):
def __init__(self, pattern, strategy, examples):
super().__init__()
self.pattern = pattern
self.strategy = strategy
self.examples = examples
# Select appropriate DSPy module based on pattern and strategy
self.processor = self.select_processor_type(pattern, strategy)
# If examples available, set as demonstrations
if examples:
self.processor.demos = examples[:5] # Use up to 5 examples
def select_processor_type(self, pattern, strategy):
"""Select processor type based on pattern"""
pattern_lower = pattern.lower()
if any(word in pattern_lower for word in ['classify', 'category', 'classification']):
return dspy.ChainOfThought("input -> reasoning, category")
elif any(word in pattern_lower for word in ['question', 'answer', 'qa']):
return dspy.ChainOfThought("question, context -> reasoning, answer")
elif any(word in pattern_lower for word in ['summary', 'summarize']):
return dspy.ChainOfThought("text -> reasoning, summary")
else:
return dspy.ChainOfThought("input -> reasoning, output")
def forward(self, **kwargs):
return self.processor(**kwargs)
return AdaptedProcessor(pattern, strategy, examples)
def calculate_adaptation_confidence(self, similar_pattern: Optional[Dict]) -> float:
"""Calculate adaptation confidence"""
if similar_pattern is None:
return 0.3 # Base confidence for new task
similarity = similar_pattern.get('similarity', 0.0)
# Calculate confidence based on similarity
confidence = min(0.3 + similarity * 0.7, 0.9)
return confidence
class FewShotLearningOptimizer:
"""Few-shot learning optimizer"""
def __init__(self):
self.example_selectors = {
'diversity': self.select_diverse_examples,
'similarity': self.select_similar_examples,
'difficulty': self.select_progressive_examples,
'uncertainty': self.select_uncertain_examples
}
def optimize_few_shot_examples(self,
candidate_examples: List[dspy.Example],
target_task: dspy.Example,
k: int = 5,
selection_strategy: str = 'diversity') -> List[dspy.Example]:
"""Optimize few-shot example selection"""
print(f"🎯 Optimizing few-shot example selection, strategy: {selection_strategy}")
if selection_strategy not in self.example_selectors:
raise ValueError(f"Unsupported selection strategy: {selection_strategy}")
selected_examples = self.example_selectors[selection_strategy](
candidate_examples, target_task, k
)
print(f"📋 Selected {len(selected_examples)} examples")
return selected_examples
def select_diverse_examples(self,
candidates: List[dspy.Example],
target: dspy.Example,
k: int) -> List[dspy.Example]:
"""Select diverse examples"""
if len(candidates) <= k:
return candidates
selected = []
remaining = candidates.copy()
# Randomly select first example
import random
first_example = random.choice(remaining)
selected.append(first_example)
remaining.remove(first_example)
# Iteratively select most diverse examples
while len(selected) < k and remaining:
best_candidate = None
best_diversity_score = -1
for candidate in remaining:
diversity_score = self.calculate_diversity_score(candidate, selected)
if diversity_score > best_diversity_score:
best_diversity_score = diversity_score
best_candidate = candidate
if best_candidate:
selected.append(best_candidate)
remaining.remove(best_candidate)
return selected
def calculate_diversity_score(self,
candidate: dspy.Example,
selected: List[dspy.Example]) -> float:
"""Calculate diversity score"""
if not selected:
return 1.0
# Simplified diversity calculation (based on text content)
candidate_text = " ".join(str(v) for v in candidate.__dict__.values())
candidate_words = set(candidate_text.lower().split())
diversity_scores = []
for selected_example in selected:
selected_text = " ".join(str(v) for v in selected_example.__dict__.values())
selected_words = set(selected_text.lower().split())
# Calculate Jaccard distance
if candidate_words and selected_words:
intersection = len(candidate_words & selected_words)
union = len(candidate_words | selected_words)
similarity = intersection / union if union > 0 else 0.0
diversity = 1.0 - similarity
else:
diversity = 1.0
diversity_scores.append(diversity)
# Return average diversity
return sum(diversity_scores) / len(diversity_scores)
def select_similar_examples(self,
candidates: List[dspy.Example],
target: dspy.Example,
k: int) -> List[dspy.Example]:
"""Select examples similar to target"""
# Calculate similarity of each candidate with target
similarity_scores = []
target_text = " ".join(str(v) for v in target.inputs().values())
target_words = set(target_text.lower().split())
for candidate in candidates:
candidate_text = " ".join(str(v) for v in candidate.inputs().values())
candidate_words = set(candidate_text.lower().split())
if target_words and candidate_words:
intersection = len(target_words & candidate_words)
union = len(target_words | candidate_words)
similarity = intersection / union if union > 0 else 0.0
else:
similarity = 0.0
similarity_scores.append((candidate, similarity))
# Sort by similarity and select top-k
similarity_scores.sort(key=lambda x: x[1], reverse=True)
return [candidate for candidate, _ in similarity_scores[:k]]
def select_progressive_examples(self,
candidates: List[dspy.Example],
target: dspy.Example,
k: int) -> List[dspy.Example]:
"""Select progressively difficult examples"""
# Estimate example difficulty
difficulty_scores = []
for candidate in candidates:
difficulty = self.estimate_example_difficulty(candidate)
difficulty_scores.append((candidate, difficulty))
# Sort by difficulty
difficulty_scores.sort(key=lambda x: x[1])
# Select progressively difficult examples
if len(difficulty_scores) <= k:
return [candidate for candidate, _ in difficulty_scores]
# Evenly distributed selection
step = len(difficulty_scores) // k
selected_indices = [i * step for i in range(k)]
return [difficulty_scores[i][0] for i in selected_indices]
def estimate_example_difficulty(self, example: dspy.Example) -> float:
"""Estimate example difficulty"""
# Simplified difficulty estimation based on text length and complexity
text_content = " ".join(str(v) for v in example.__dict__.values())
# Length factor
length_factor = len(text_content) / 1000.0 # Normalized
# Vocabulary complexity factor
words = text_content.split()
unique_words = len(set(words))
vocabulary_complexity = unique_words / len(words) if words else 0.0
# Syntactic complexity factor (simplified)
sentence_complexity = text_content.count('.') + text_content.count('?') + text_content.count('!')
sentence_complexity = sentence_complexity / 10.0 # Normalized
# Combined difficulty score
difficulty = (length_factor * 0.3 +
vocabulary_complexity * 0.4 +
sentence_complexity * 0.3)
return min(difficulty, 1.0)
def select_uncertain_examples(self,
candidates: List[dspy.Example],
target: dspy.Example,
k: int) -> List[dspy.Example]:
"""Select high uncertainty examples"""
# This requires a pretrained model to estimate uncertainty
# Simplified implementation: based on example "uniqueness"
uncertainty_scores = []
for candidate in candidates:
uncertainty = self.estimate_example_uncertainty(candidate, candidates)
uncertainty_scores.append((candidate, uncertainty))
# Select examples with highest uncertainty
uncertainty_scores.sort(key=lambda x: x[1], reverse=True)
return [candidate for candidate, _ in uncertainty_scores[:k]]
def estimate_example_uncertainty(self,
example: dspy.Example,
all_examples: List[dspy.Example]) -> float:
"""Estimate example uncertainty"""
# Based on "uniqueness" of example in entire set
example_text = " ".join(str(v) for v in example.__dict__.values())
example_words = set(example_text.lower().split())
similarities = []
for other_example in all_examples:
if other_example == example:
continue
other_text = " ".join(str(v) for v in other_example.__dict__.values())
other_words = set(other_text.lower().split())
if example_words and other_words:
intersection = len(example_words & other_words)
union = len(example_words | other_words)
similarity = intersection / union if union > 0 else 0.0
similarities.append(similarity)
# Uncertainty inversely related to similarity
avg_similarity = sum(similarities) / len(similarities) if similarities else 0.0
uncertainty = 1.0 - avg_similarity
return uncertainty
# Usage examples
def demonstrate_meta_learning():
"""Demonstrate meta-learning"""
meta_learner = MetaLearner()
# Create multiple learning tasks
tasks = [
{
'name': 'Sentiment Analysis',
'type': 'classification',
'examples': [
dspy.Example(text="This movie is amazing", label="positive").with_inputs('text'),
dspy.Example(text="This movie is terrible", label="negative").with_inputs('text'),
]
},
{
'name': 'Topic Classification',
'type': 'classification',
'examples': [
dspy.Example(text="Stock market rises today", label="finance").with_inputs('text'),
dspy.Example(text="New phone launched", label="technology").with_inputs('text'),
]
}
]
# Meta-learning process
meta_learning_result = meta_learner.learn_from_tasks(tasks)
print(f"\n📊 Meta-learning results:")
print(f"Common patterns: {meta_learning_result['common_patterns']}")
print(f"Meta insights: {meta_learning_result['meta_insights']}")
# Adapt to new task
new_task_examples = [
dspy.Example(text="This restaurant has good service", rating="4").with_inputs('text'),
dspy.Example(text="This restaurant has a nice environment", rating="5").with_inputs('text'),
]
adaptation_result = meta_learner.adapt_to_new_task(
new_task_examples,
task_type_hint='classification'
)
print(f"\n🎯 New task adaptation:")
print(f"Confidence: {adaptation_result['confidence']:.2f}")
print(f"Task pattern: {adaptation_result['task_pattern'][:100]}...")
return meta_learner, adaptation_result
# demo_meta_learning = demonstrate_meta_learning()[Continue with remaining sections translated in same format…]
Through this chapter, you should have mastered advanced patterns and techniques in DSPy. These technologies represent cutting-edge directions in AI programming and can help you build more intelligent, adaptive, and reliable AI application systems. In practical applications, select appropriate technology combinations based on specific needs.