dspy-2-modules

name	dspy-2-modules
description	Sub-skill of dspy: 2. Modules.
version	1.0.0
category	ai-prompting
type	reference
scripts_exempt	true

2. Modules

2. Modules

ChainOfThought for Complex Reasoning:

class TechnicalQA(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 answering
class TechnicalExpert(dspy.Module):
    def __init__(self):
        super().__init__()
        self.answer_question = dspy.ChainOfThought(TechnicalQA)

    def forward(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}")

Multi-Stage Pipeline Module:

class DocumentSummary(dspy.Signature):
    """Summarize a technical document."""
    document = dspy.InputField()
    summary = dspy.OutputField()

class KeyPointExtraction(dspy.Signature):
    """Extract key points from a summary."""
    summary = dspy.InputField()
    key_points = dspy.OutputField(desc="List of 3-5 key points")

class ActionItemGeneration(dspy.Signature):
    """Generate action items from key points."""
    key_points = dspy.InputField()
    action_items = dspy.OutputField(desc="List of actionable next steps")

class DocumentProcessor(dspy.Module):
    """Multi-stage document processing pipeline."""

    def __init__(self):
        super().__init__()
        self.summarize = dspy.ChainOfThought(DocumentSummary)
        self.extract_points = dspy.Predict(KeyPointExtraction)
        self.generate_actions = dspy.Predict(ActionItemGeneration)

    def forward(self, document):
        # Stage 1: Summarize
        summary_result = self.summarize(document=document)

        # Stage 2: Extract key points
        points_result = self.extract_points(summary=summary_result.summary)

        # Stage 3: Generate actions
        actions_result = self.generate_actions(key_points=points_result.key_points)

        return dspy.Prediction(
            summary=summary_result.summary,
            key_points=points_result.key_points,
            action_items=actions_result.action_items
        )

# Usage
processor = DocumentProcessor()
result = processor(document="[Long engineering document text...]")

()
()
()

ReAct Module for Tool Use:

class CalculateTension(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")

class SearchStandards(dspy.Signature):
    """Search engineering standards database."""
    query = dspy.InputField(desc="Search query")
    standards = dspy.OutputField(desc="Relevant standards found")

class EngineeringReActAgent(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]
        )

    def calculate_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)
        return f"Estimated tension: {tension:.1f} kN"

    def () -> :
        
        standards_db = {
            : [, , ],
            : [, ],
            : [, ]
        }
         key, value  standards_db.items():
             key  query.lower():
                 
         

     ():
         .react(question=question)


agent = EngineeringReActAgent()
result = agent(
    question=
)
(result.answer)