Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.
Quelldateien prüfen
Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
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)
