import dspy
# Configure LLM
lm = dspy.OpenAI(model="gpt-4", max_tokens=1000)
dspy.settings.configure(lm=lm)
# Inline signature (simple)
classify = dspy.Predict("document -> category")
result = classify(document="The mooring line tension exceeded limits.")
print(result.category)
# Class-based signature (recommended)classSentimentAnalysis(dspy.Signature):
"""Analyze the sentiment of engineering feedback."""
feedback = dspy.InputField(desc="Engineering feedback or review text")
sentiment = dspy.OutputField(desc="Sentiment: positive, negative, or neutral")
confidence = dspy.OutputField(desc="Confidence score 0-1")
# Use signature
analyzer = dspy.Predict(SentimentAnalysis)
result = analyzer(feedback="The mooring design passed all safety checks.")
print(f"Sentiment: {result.sentiment}, Confidence: {result.confidence}")
Complex Signatures with Multiple Fields:
classEngineeringAnalysis(dspy.Signature):
"""Analyze an engineering report and extract key insights."""
report_text = dspy.InputField(
desc="Full text of the engineering report"
)
domain = dspy.InputField(
desc="Engineering domain (offshore, structural, mechanical)"
)
summary = dspy.OutputField(
desc="Concise 2-3 sentence summary of findings"
)
key_metrics = dspy.OutputField(
desc="List of key metrics mentioned with values"
)
risk_factors = dspy.OutputField(
desc="Identified risk factors and concerns"
)
recommendations = dspy.OutputField(
desc="Actionable recommendations from the report"
)
confidence_level = dspy.OutputField(
desc="Overall confidence in analysis: high, medium, or low"
)
# Create predictor
report_analyzer = dspy.Predict(EngineeringAnalysis)
# Analyze report
result = report_analyzer(
report_text="""
The mooring analysis for Platform Alpha shows maximum tensions
of 2,450 kN under 100-year storm conditions. Safety factors
range from 1.72 to 2.15 across all lines. Line 3 shows the
lowest margin at the fairlead connection. Fatigue life estimates
indicate 35-year service life, exceeding the 25-year requirement.
Chain wear measurements show 8% diameter loss after 5 years.
""",
domain="offshore"
)
print(f"Summary: {result.summary}")
print(f"Key Metrics: {result.key_metrics}")
print(f"Risk Factors: {result.risk_factors}")
print(f"Recommendations: {result.recommendations}")
2. Modules
ChainOfThought for Complex Reasoning:
classTechnicalQA(dspy.Signature):
"""Answer technical engineering questions with reasoning."""
context = dspy.InputField(desc="Technical context and background")
question = dspy.InputField(desc="Technical question to answer")
answer = dspy.OutputField(desc="Detailed technical answer")
# ChainOfThought adds reasoning before answeringclassTechnicalExpert(dspy.Module):
def__init__(self):
super().__init__()
self.answer_question = dspy.ChainOfThought(TechnicalQA)
defforward(self, context, question):
result = self.answer_question(context=context, question=question)
return result
# Usage
expert = TechnicalExpert()
result = expert(
context="""
Catenary mooring systems use the weight of the chain to provide
restoring force. The touchdown point moves as the vessel offsets.
Line tension is a function of the catenary geometry and pretension.
""",
question="How does water depth affect mooring line tension?"
)
print(f"Reasoning: {result.rationale}")
print(f"Answer: {result.answer}")
classCalculateTension(dspy.Signature):
"""Calculate mooring line tension."""
depth = dspy.InputField(desc="Water depth in meters")
line_length = dspy.InputField(desc="Line length in meters")
pretension = dspy.InputField(desc="Pretension in kN")
result = dspy.OutputField(desc="Tension calculation result")
classSearchStandards(dspy.Signature):
"""Search engineering standards database."""
query = dspy.InputField(desc="Search query")
standards = dspy.OutputField(desc="Relevant standards found")
classEngineeringReActAgent(dspy.Module):
"""Agent that can reason and act using tools."""def__init__(self):
super().__init__()
self.react = dspy.ReAct(
signature="question -> answer",
tools=[self.calculate_tension, self.search_standards]
)
defcalculate_tension(self, depth: float, line_length: float, pretension: float) -> str:
"""Calculate approximate mooring line tension."""import math
suspended = math.sqrt(line_length**2 - depth**2)
tension = pretension * (1 + depth / suspended * 0.1)
returnf"Estimated tension: {tension:.1f} kN"defsearch_standards(self, query: str) -> str:
"""Search for relevant engineering standards."""
standards_db = {
"mooring": ["API RP 2SK", "DNV-OS-E301", "ISO 19901-7"],
"fatigue": ["DNV-RP-C203", "API RP 2A-WSD"],
"structural": ["AISC 360", "API RP 2A-WSD"]
}
for key, value in standards_db.items():
if key in query.lower():
returnf"Relevant standards: {', '.join(value)}"return"No specific standards found for query"defforward(self, question):
returnself.react(question=question)
# Usage
agent = EngineeringReActAgent()
result = agent(
question="What is the tension for a 350m line in 100m depth with 500kN pretension?"
)
print(result.answer)
3. Retrieval-Augmented Generation
RAG with DSPy:
import dspy
from dspy.retrieve.chromadb_rm import ChromadbRM
# Configure retriever
retriever = ChromadbRM(
collection_name="engineering_docs",
persist_directory="./chroma_db",
k=5
)
# Configure DSPy with retriever
dspy.settings.configure(
lm=dspy.OpenAI(model="gpt-4"),
rm=retriever
)
classRAGSignature(dspy.Signature):
"""Answer questions using retrieved context."""
context = dspy.InputField(desc="Retrieved relevant passages")
question = dspy.InputField(desc="Question to answer")
answer = dspy.OutputField(desc="Answer based on context")
classRAGModule(dspy.Module):
"""RAG module with retrieval and generation."""def__init__(self, num_passages=5):
super().__init__()
self.retrieve = dspy.Retrieve(k=num_passages)
self.generate = dspy.ChainOfThought(RAGSignature)
defforward(self, question):
# Retrieve relevant passages
passages = self.retrieve(question).passages
# Generate answer with context
context = "\n\n".join(passages)
result = self.generate(context=context, question=question)
return dspy.Prediction(
answer=result.answer,
passages=passages,
reasoning=result.rationale
)
# Usage
rag = RAGModule(num_passages=5)
result = rag(question="What are the safety factor requirements for moorings?")
print(f"Answer: {result.answer}")
print(f"Sources: {len(result.passages)} passages retrieved")
Multi-Hop RAG:
classMultiHopRAG(dspy.Module):
"""
Multi-hop RAG that retrieves, reasons, and retrieves again
for complex questions requiring multiple pieces of information.
"""def__init__(self, num_hops=2, passages_per_hop=3):
super().__init__()
self.num_hops = num_hops
self.retrieve = dspy.Retrieve(k=passages_per_hop)
self.generate_query = dspy.ChainOfThought(
"context, question -> search_query"
)
self.generate_answer = dspy.ChainOfThought(RAGSignature)
defforward(self, question):
context = []
current_query = question
for hop inrange(self.num_hops):
# Retrieve for current query
passages = self.retrieve(current_query).passages
context.extend(passages)
if hop < self.num_hops - 1:
# Generate refined query for next hop
all_context = "\n\n".join(context)
query_result = self.generate_query(
context=all_context,
question=question
)
current_query = query_result.search_query
# Final answer generation
full_context = "\n\n".join(context)
result = self.generate_answer(
context=full_context,
question=question
)
return dspy.Prediction(
answer=result.answer,
hops=self.num_hops,
total_passages=len(context)
)
# Usage
multi_hop_rag = MultiHopRAG(num_hops=3, passages_per_hop=3)
result = multi_hop_rag(
question="How does fatigue analysis relate to mooring safety factors?"
)
4. Optimizers
BootstrapFewShot Optimizer:
from dspy.teleprompt import BootstrapFewShot
classClassifyReport(dspy.Signature):
"""Classify engineering report type."""
report_text = dspy.InputField()
report_type = dspy.OutputField(
desc="Type: analysis, inspection, design, or incident"
)
classReportClassifier(dspy.Module):
def__init__(self):
super().__init__()
self.classify = dspy.Predict(ClassifyReport)
defforward(self, report_text):
returnself.classify(report_text=report_text)
# Create training data
trainset = [
dspy.Example(
report_text="The mooring analysis shows maximum tensions...",
report_type="analysis"
).with_inputs("report_text"),
dspy.Example(
report_text="Visual inspection of Line 3 revealed corrosion...",
report_type="inspection"
).with_inputs("report_text"),
dspy.Example(
report_text="The new platform design incorporates...",
report_type="design"
).with_inputs("report_text"),
dspy.Example(
report_text="At 14:32, the vessel experienced sudden offset...",
report_type="incident"
).with_inputs("report_text"),
# Add more examples...
]
# Define metricdefclassification_accuracy(example, prediction, trace=None):
return example.report_type.lower() == prediction.report_type.lower()
# Optimize
optimizer = BootstrapFewShot(
metric=classification_accuracy,
max_bootstrapped_demos=4,
max_labeled_demos=8
)
# Compile optimized module
optimized_classifier = optimizer.compile(
ReportClassifier(),
trainset=trainset
)
# Use optimized classifier
result = optimized_classifier(
report_text="Fatigue analysis indicates remaining life of 15 years..."
)
print(f"Type: {result.report_type}")
BootstrapFewShotWithRandomSearch:
from dspy.teleprompt import BootstrapFewShotWithRandomSearch
# More thorough optimization with search
optimizer = BootstrapFewShotWithRandomSearch(
metric=classification_accuracy,
max_bootstrapped_demos=4,
max_labeled_demos=8,
num_candidate_programs=10,
num_threads=4
)
# This searches for the best combination of examples
optimized = optimizer.compile(
ReportClassifier(),
trainset=trainset,
valset=valset # Optional validation set
)
# Increase number of training examples# Ensure diverse, high-quality examples# Try different optimizer settings
optimizer = BootstrapFewShotWithRandomSearch(
metric=metric,
max_bootstrapped_demos=8, # Increase
num_candidate_programs=20, # More search
num_threads=8
)
Module Too Slow
# Use faster model for compilation
compile_lm = dspy.OpenAI(model="gpt-4.1-mini")
deploy_lm = dspy.OpenAI(model="gpt-4")
with dspy.settings.context(lm=compile_lm):
optimized = optimizer.compile(module, trainset=data)
# Deploy with stronger model
dspy.settings.configure(lm=deploy_lm)
Out of Memory
# Process in batches
batch_size = 10for i inrange(0, len(trainset), batch_size):
batch = trainset[i:i+batch_size]
process_batch(batch)
