Shift Your Paradigm: Building Self-Improving LLM Workflows with DSPy

@article{Khattabetal2022,
  author  = {Khattab, Omar and Santhanam, Keshav and Li, Xiang Lisa and Hall, David and Liang, Percy and Potts, Christopher and Zaharia, Matei},
  title   = {Demonstrate-Search-Predict: Composing retrieval and language models for knowledge-intensive NLP},
  journal = {arXiv},
  year    = {2022},
  doi     = {10.48550/arxiv.2212.14024}
}

@article{Khattabetal2023,
  author  = {Khattab, Omar and Santhanam, Keshav and Li, Xiang Lisa and Hall, David and Liang, Percy and Potts, Christopher and Zaharia, Matei},
  title   = {DSPy: Compiling Declarative Language Model Calls into Self-Improving Pipelines},
  journal = {OpenReview},
  year    = {2023},
  url     = {https://openreview.net/forum?id=sY5N0zY5Od}
}

import dspy
from typing import List

# Configure the global language model setting
lm = dspy.LM("openai/gpt-4o-mini", api_key="your-api-key")
dspy.settings.configure(lm=lm)

# Define a Class-Based Signature for clean, type-hinted parsing
class IntentSignature(dspy.Signature):
    """Classify the incoming customer service message into an explicit category."""
    message: str = dspy.InputField()
    labels: List[str] = dspy.InputField()
    intent_label: str = dspy.OutputField()

# Instantiate a Chain of Thought predictive module using the signature
classifier = dspy.ChainOfThought(IntentSignature)

# Execute the module cleanly like a standard Python function
response = classifier(
    message="I deserve a refund for the frozen app.",
    labels=["Subscription Cancel", "Refund Request", "Bug Report"]
)
print(f"Reasoning: {response.rationale}")
print(f"Predicted Intent: {response.intent_label}")

# Constructing explicit training/validation data structures
examples = [
    dspy.Example(
        message="I want to end my subscription", 
        labels=["Cancel Subscription", "Refund Request"], 
        intent_label="Cancel Subscription"
    ).with_inputs("message", "labels"),
    dspy.Example(
        message="The screen is frozen", 
        labels=["Cancel Subscription", "Bug Report"], 
        intent_label="Bug Report"
    ).with_inputs("message", "labels")
]

# Defining a functional metric to check accuracy
def validate_intent_metric(example, prediction, trace=None):
    return example.intent_label == prediction.intent_label

# DSPy's built-in evaluation harness for executing batch test simulations
from dspy.evaluate import Evaluate

evaluator = Evaluate(devset=examples, metric=validate_intent_metric, num_threads=4)
# evaluator(classifier) returns structured evaluation matrices

from dspy.teleprompt import BootstrapFewShot

# Define the compilation configuration setup
optimizer = BootstrapFewShot(
    metric=validate_intent_metric,
    max_bootstrapped_demos=2,
    max_labeled_demos=2
)

# Compile transforms the unoptimized module into a hyper-optimized program
optimized_classifier = optimizer.compile(student=classifier, trainset=exam

import dspy
from pydantic import BaseModel, Field, ValidationError
import json

lm = dspy.LM("openai/gpt-4o-mini", api_key="sk-your-key-here", temperature=0.0)
dspy.settings.configure(lm=lm)

# 1. Define target downstream schema validation object
class UserProfileSchema(BaseModel):
    user_id: int = Field(..., description="Unique integer ID")
    email: str = Field(..., description="Valid standard email address string")
    account_status: str = Field(..., description="Must be exactly 'active', 'suspended', or 'pending'")

# 2. Define DSPy Signatures for generation and dynamic error correction
class ExtractDataSignature(dspy.Signature):
    """Extract raw input text and convert it into a valid formatted JSON object matching the target schema."""
    raw_text: str = dspy.InputField()
    target_schema_description: str = dspy.InputField()
    extracted_json: str = dspy.OutputField(desc="Raw parsable JSON block containing extracted keys")

class SelfCorrectSignature(dspy.Signature):
    """Examine prior malformed JSON output alongside validation errors to emit a structurally corrected JSON block."""
    malformed_json: str = dspy.InputField()
    validation_error_msg: str = dspy.InputField()
    corrected_json: str = dspy.OutputField(desc="Cleaned, valid, schema-compliant JSON object")

# 3. Create a Modular, Multi-Hop Program with an explicit internal feedback loop
class GuardrailDataExtractor(dspy.Module):
    def __init__(self):
        super().__init__()
        self.extractor = dspy.Predict(ExtractDataSignature)
        self.corrector = dspy.ChainOfThought(SelfCorrectSignature)

    def forward(self, raw_text: str):
        schema_desc = (
            "JSON fields required: user_id (int), email (str), "
            "account_status (string literal matching 'active', 'suspended', 'pending')"
        )

        # Step 1: Initial parsing extraction execution attempt
        pred = self.extractor(raw_text=raw_text, target_schema_description=schema_desc)
        extracted_text = pred.extracted_json

        # Clean potential markdown wrapping artifacts if emitted by basic models
        if "```

json" in extracted_text:
            extracted_text = extracted_text.split("

```json")[1].split("```

")[0].strip()

        # Step 2: Validate against Pydantic constraint layer
        try:
            parsed_data = json.loads(extracted_text)
            UserProfileSchema(**parsed_data)
            return dspy.Prediction(valid=True, data=parsed_data, logs="Success on initial pass.")
        except (ValidationError, json.JSONDecodeError) as e:
            error_feedback = str(e)

            # Step 3: Self-correction trigger branch passing runtime feedback traces to a CoT corrector module
            correction_pred = self.corrector(
                malformed_json=extracted_text,
                validation_error_msg=error_feedback
            )

            corrected_text = correction_pred.corrected_json
            if "

```json" in corrected_text:
                corrected_text = corrected_text.split("```

json")[1].split("

```")[0].strip()

            try:
                final_data = json.loads(corrected_text)
                UserProfileSchema(**final_data)
                return dspy.Prediction(valid=True, data=final_data, logs=f"Corrected after failure. Error was: {error_feedback}")
            except Exception as final_fail:
                return dspy.Prediction(valid=False, data=corrected_text, logs=f"Self-correction collapsed: {str(final_fail)}")

# --- Execution Simulation ---
guardrail_pipeline = GuardrailDataExtractor()

# Intentionally noisy input designed to cause parsing errors (invalid email format and illegal account state string)
noisy_input_string = (
    "Customer record update log: The individual with id number 9942 has an address at 'john_at_domain_com'. "
    "We need to flag his profile as initialized immediately."
)

result = guardrail_pipeline(raw_text=noisy_input_string)
print("\n--- Guardrail Program Output Evaluation Matrix ---")
print(f"Validation Status Verified: {result.valid}")
print(f"Final Parsed Safe Record Output: {json.dumps(result.data, indent=2)}")
print(f"Internal Pipeline Processing Logs: {result.logs}")

pip install dspy-ai transformers torch

import os
import dspy
from typing import List

# 1. Initialize the IBM Granite Model from Hugging Face via DSPy
# We use the official 'ibm-granite/granite-3.0-8b-instruct' model.
# Ensure your HF_TOKEN environment variable is set if using gated weights.
os.environ["HF_TOKEN"] = "your_huggingface_token_here"

print("Loading IBM Granite model from Hugging Face...")
granite_llm = dspy.HFClientTGI(
    model="ibm-granite/granite-3.0-8b-instruct", 
    port=None,  
    temperature=0.2
)

# Configure DSPy globally to use Granite as the default Language Model
dspy.settings.configure(lm=granite_llm)


# 2. Define the Agent's Signature (The Behavioral System Contract)
class ResearchAgentSignature(dspy.Signature):
    """
    An advanced multi-hop research agent designed to answer complex user queries 
    by analyzing provided context chunks. The agent must declare its step-by-step 
    rationale before formulating a synthesized, factual final response.
    """
    user_query: str = dspy.InputField(desc="The multi-part question or task the user wants resolved.")
    retrieved_context: str = dspy.InputField(desc="Relevant contextual passages, documentation, or search results.")

    agent_thought: str = dspy.OutputField(desc="Internal analytical chain-of-thought, reasoning steps, or gap analysis.")
    final_response: str = dspy.OutputField(desc="The concrete, verified final answer to the user query.")


# 3. Build the Agentic Module with Self-Correction Loops
class GraniteResearchAgent(dspy.Module):
    def __init__(self):
        super().__init__()
        # Using ChainOfThought forces Granite to output the 'agent_thought' field before the final response
        self.research_step = dspy.ChainOfThought(ResearchAgentSignature)

    def forward(self, query: str, context_documents: List[str]):
        # Format list of documents into a single readable text block
        formatted_context = "\n".join([f"-[Doc {i+1}]: {doc}" for i, doc in enumerate(context_documents)])

        current_query = query
        max_attempts = 2

        for attempt in range(max_attempts):
            # Execute the core LLM call using the compiled/optimized signature
            prediction = self.research_step(
                user_query=current_query, 
                retrieved_context=formatted_context
            )

            # Simple Agentic Self-Correction Rule:
            # If the model discovers it lacks enough context in its thought process, 
            # we modify the query dynamically to perform a broader internal pass.
            if "insufficient context" in prediction.agent_thought.lower() and attempt == 0:
                print(f"⚠️ [Agent Feedback Loop]: Granite detected missing info. Adjusting strategy...")
                current_query = f"{query} (Provide a best-effort answer using generalized structural reasoning if direct facts are sparse)."
                continue

            return dspy.Prediction(
                rationale=prediction.agent_thought,
                answer=prediction.final_response
            )

# 4. Simulation / Execution
if __name__ == "__main__":
    # Mocking a vector database retrieval output for the context input
    sample_knowledge_base = [
        "IBM Granite 3.0 models are architectures trained on over 12 trillion tokens of high-quality data.",
        "Granite 3.0 Instruct models show significant enhancements in safety, enterprise tool governance, and RAG architectures.",
        "DSPy allows users to separate pipeline structure from prompt engineering choices by treating prompts as parameters."
    ]

    # Instantiate the agent
    agent = GraniteResearchAgent()

    # Run a complex query
    user_question = "How does IBM Granite 3.0's architecture fit into programmatic framework environments like DSPy?"

    print(f"\n🚀 Dispatching Agent with query: '{user_question}'\n")
    output = agent(query=user_question, context_documents=sample_knowledge_base)

    # Output the results
    print("================ AGENT REASONING STEP ================")
    print(output.rationale)
    print("\n================== FINAL RESPONSE ==================")
    print(output.answer)