Dspy Utils

DSPy Utilities: Caching, Debugging, Save/Load, and Validation

Guide the user through DSPy's utility functions -- controlling caching, debugging calls, persisting optimized programs, and enforcing runtime constraints with reward functions.

Looking for streaming, async, or MCP? These have dedicated skills now:

- Streaming tokens to a UI -- see /dspy-streaming

- Async execution and FastAPI -- see /dspy-async

- MCP server integration -- see /dspy-mcp

Step 1: Which utility do you need?

Ask the user before diving in:

1. What are you trying to do? Debug a failing program, save/load an optimized program, control caching, or validate outputs with reward functions? 2. Is this for development or production? Development needs (debugging, cache control) differ from production needs (save/load, validation).

Then jump to the relevant section below.

2. configure_cache -- controlling cache behavior

DSPy caches LM responses by default to reduce costs and speed up development. Use dspy.configure_cache to control this globally.

# Disable caching entirely
dspy.configure_cache(enable=False)

# Re-enable caching
dspy.configure_cache(enable=True)

# DSPy 3.2+ - harden the on-disk cache against untrusted pickle payloads
dspy.configure_cache(restrict_pickle=True)

Per-LM cache control

You can also control caching per LM instance:

# This LM never caches
lm_no_cache = dspy.LM("openai/gpt-4o-mini", cache=False)

# This LM caches (default)
lm_cached = dspy.LM("openai/gpt-4o-mini", cache=True)

When to disable caching

• Generating diverse outputs -- when you need different responses for the same prompt (e.g., data generation)
• Testing real latency -- cache hits are instant, which skews benchmarks
• Streaming -- caching may interfere with streaming behavior in some configurations

Cache is stored locally on disk. Identical calls (same prompt, parameters, model) return cached results with no API call.

When NOT to disable caching: During optimization runs -- optimizers rely heavily on cache to avoid redundant LM calls. Disabling cache globally during optimization dramatically increases cost and time.

3. inspect_history -- debugging LM calls

dspy.inspect_history shows the raw prompts and responses from recent LM calls. This is the single most useful debugging tool in DSPy.

import dspy

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

classify = dspy.Predict("text -> label")
classify(text="Great product!")

# See what was actually sent to and received from the LM
dspy.inspect_history(n=1)  # Show last 1 call
dspy.inspect_history(n=3)  # Show last 3 calls

What inspect_history shows

• The full prompt sent to the LM (including system message, few-shot demos, instructions)
• The raw LM response
• Which adapter formatted the prompt (ChatAdapter, JSONAdapter, etc.)

Debugging workflow

1. Run your program on a failing input 2. Call dspy.inspect_history(n=1) to see the last LM call 3. Check if the prompt makes sense -- are the instructions clear? Are few-shot demos relevant? 4. Check the raw response -- did the LM follow the format? Did it hallucinate? 5. Adjust your signature, module, or optimization strategy based on what you see

Verbose logging

For more detailed tracing, configure DSPy with an empty trace list:

dspy.configure(lm=lm, trace=[])

You can also print a module to see its structure:

print(my_program)  # Shows module tree with all sub-modules and signatures

4. save/load -- persisting optimized programs

After optimizing a DSPy program, save its learned state (few-shot demos, instructions) for production use.

Save

# After optimization
optimized = optimizer.compile(my_program, trainset=trainset)
optimized.save("optimized_program.json")

Load

# In production -- create a fresh instance, then load state
program = MyProgram()
program.load("optimized_program.json")

# Use it
result = program(question="What is DSPy?")

What gets saved

• Few-shot demonstrations discovered by optimizers
• Optimized instructions (from MIPROv2, GEPA, etc.)
• Any state tracked by dspy.Predict modules

What does NOT get saved

• Python logic in forward() -- that's your code, it must exist at load time
• Model weights (unless you used BootstrapFinetune)
• LM configuration -- you must call dspy.configure() before loading

Production deployment pattern

import dspy

class MyPipeline(dspy.Module):
    def __init__(self):
        self.classify = dspy.Predict("text -> category")
        self.respond = dspy.ChainOfThought("text, category -> response")

    def forward(self, text):
        cat = self.classify(text=text)
        return self.respond(text=text, category=cat.category)

# --- Optimization (run once) ---
# optimizer = dspy.MIPROv2(metric=metric, auto="medium")
# optimized = optimizer.compile(MyPipeline(), trainset=trainset)
# optimized.save("pipeline_v1.json")

# --- Production (run on every request) ---
lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

pipeline = MyPipeline()
pipeline.load("pipeline_v1.json")

result = pipeline(text="How do I reset my password?")

5. dspy.Refine and dspy.BestOfN -- reward-based output validation

Use dspy.Refine to wrap any module and retry until a reward function returns a score meeting a threshold. This replaced dspy.Assert/dspy.Suggest in DSPy 3.x:

import dspy

qa = dspy.ChainOfThought("question -> answer")

def answer_reward(args, pred):
    """Score answer quality. Returns 0.0-1.0."""
    if not pred.answer.strip():
        return 0.0
    if len(pred.answer.split()) < 5:
        return 0.5  # soft penalty for short answers
    return 1.0

validated_qa = dspy.Refine(
    module=qa,
    N=3,
    reward_fn=answer_reward,
    threshold=1.0,
)

result = validated_qa(question="What is DSPy?")

• `dspy.Refine` -- retries with feedback from the reward function until threshold is met or N attempts exhausted. Use when later attempts can improve based on earlier failures.
• `dspy.BestOfN` -- runs N independent attempts and returns the best-scoring one. Use when attempts are independent and cross-attempt feedback would not help.

For detailed patterns and examples, see `/dspy-refine` and `/dspy-best-of-n`.

Gotchas

1. `save()` does not persist `forward()` logic -- only learned state (demos, instructions) is saved. The class definition must exist in your production code at load time. 2. Must `dspy.configure()` before `load()` -- loading a saved program before configuring the LM causes silent failures where the program runs but uses no LM (or the wrong one). 3. `inspect_history` shows cached calls too -- after a cache hit, inspect_history still shows the call, but the prompt may look different from what was originally sent. Disable cache if you need exact prompt inspection. 4. Claude disables caching during optimization. Do NOT disable cache globally during optimizer runs -- optimizers rely heavily on cache to avoid redundant LM calls. Disabling cache during optimization dramatically increases cost and time.

Additional resources

• DSPy saving/loading guide
• For API details, see reference.md
• For worked examples, see examples.md

Cross-references

Install any skill: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill <name>

• Streaming tokens to a UI -- see /dspy-streaming
• Async execution and FastAPI -- see /dspy-async
• MCP server integration -- see /dspy-mcp
• `/dspy-lm` -- Configure language models, per-LM caching, inspect_history on LM instances
• `/dspy-modules` -- Build composable programs with dspy.Module, save/load patterns
• `/ai-tracing-requests` -- Production observability and tracing for DSPy programs
• `/dspy-refine` -- Refine patterns, reward functions, and iterative improvement
• `/dspy-best-of-n` -- BestOfN for independent sampling without cross-attempt feedback
• `/ai-serving-apis` -- Serve DSPy programs as web APIs
• Install `/ai-do` if you do not have it -- it routes any AI problem to the right skill and is the fastest way to work: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill ai-do

Examples: DSPy Utilities

Example 1: Streaming responses in a web app

Stream a DSPy program's output through a FastAPI endpoint using streamify and StreamListener. The user sees tokens arrive incrementally instead of waiting for the full response.

import dspy
from dspy.streaming import streamify, StreamListener
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

# Configure DSPy
lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

# Define the program
class QA(dspy.Module):
    def __init__(self):
        self.answer = dspy.ChainOfThought("question -> answer")

    def forward(self, question):
        return self.answer(question=question)

qa = QA()

@app.get("/ask")
async def ask(question: str):
    # Create a fresh listener per request
    answer_listener = StreamListener(signature_field_name="answer")

    # Wrap the program for streaming
    streaming_qa = streamify(
        qa,
        stream_listeners=[answer_listener],
        include_final_prediction_in_output_stream=False,
    )

    async def generate():
        async for chunk in streaming_qa(question=question):
            # Each chunk has the streamed field as an attribute
            if hasattr(chunk, "answer"):
                yield chunk.answer

    return StreamingResponse(generate(), media_type="text/plain")

What to notice

• StreamListener is created fresh per request -- do not reuse listeners across requests unless you set allow_reuse=True.
• include_final_prediction_in_output_stream=False prevents the final Prediction object from appearing in the stream, since we only want the incremental text.
• The streamify wrapper returns an async generator, which maps naturally to FastAPI's StreamingResponse.
• The listener internally buffers ~10 tokens to detect field boundary delimiters. The first few tokens may arrive with a slight delay.

Example 2: Debug workflow with inspect_history

Walk through a typical debugging session when a DSPy program produces unexpected output. Use inspect_history to see exactly what the LM received and returned.

import dspy
from typing import Literal

# Set up
lm = dspy.LM("openai/gpt-4o-mini", temperature=0.0)  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

# A classification program that isn't working right
class TicketClassifier(dspy.Module):
    def __init__(self):
        self.classify = dspy.Predict(
            "ticket_text -> urgency: Literal['low', 'medium', 'high', 'critical']"
        )

    def forward(self, ticket_text):
        return self.classify(ticket_text=ticket_text)

classifier = TicketClassifier()

# --- Step 1: Run on a failing input ---
result = classifier(ticket_text="Production database is down, all customers affected")
print(f"Got: {result.urgency}")
# Expected: "critical", got: "high" -- why?

# --- Step 2: Inspect what was sent to the LM ---
dspy.inspect_history(n=1)
# This prints:
#   - The full system prompt DSPy generated
#   - The user message with the ticket text
#   - The raw LM response
#   - Which adapter formatted the request

# --- Step 3: Look at the prompt ---
# You might see that the signature description is too vague.
# The LM doesn't know that "all customers affected" implies critical urgency.

# --- Step 4: Improve the signature with a better docstring ---
class ClassifyTicket(dspy.Signature):
    """Classify support ticket urgency. Critical means production outages
    affecting multiple customers. High means significant issues affecting
    individual users."""
    ticket_text: str = dspy.InputField()
    urgency: Literal["low", "medium", "high", "critical"] = dspy.OutputField()

class BetterClassifier(dspy.Module):
    def __init__(self):
        self.classify = dspy.Predict(ClassifyTicket)

    def forward(self, ticket_text):
        return self.classify(ticket_text=ticket_text)

better = BetterClassifier()
result = better(ticket_text="Production database is down, all customers affected")
print(f"Got: {result.urgency}")

# --- Step 5: Inspect again to verify the improved prompt ---
dspy.inspect_history(n=1)
# Now the prompt includes the detailed docstring, and the LM returns "critical"

# --- Bonus: Print the module tree to verify structure ---
print(better)
# BetterClassifier(
#   classify = Predict(ClassifyTicket)
# )

What to notice

• dspy.inspect_history(n=1) is your first move when a program behaves unexpectedly. It shows the exact prompt the LM saw.
• The most common fixes after inspecting: improve the signature docstring, add desc= to InputField/OutputField, or add few-shot examples via optimization.
• Use print(module) to verify the module structure -- sometimes a sub-module is wired incorrectly.
• Set temperature=0.0 during debugging for deterministic, reproducible outputs.

Example 3: Save/load optimized programs for production deployment

Optimize a program once, save the learned state, and load it in production without re-running optimization.

import dspy
from dspy.evaluate import Evaluate

# ==============================
# Part A: Optimize and save (run once, offline)
# ==============================

lm = dspy.LM("openai/gpt-4o-mini", temperature=0.0)  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

# Define the program
class SupportResponder(dspy.Module):
    def __init__(self):
        self.classify = dspy.Predict("ticket -> category, urgency")
        self.respond = dspy.ChainOfThought("ticket, category, urgency -> response")

    def forward(self, ticket):
        classification = self.classify(ticket=ticket)
        return self.respond(
            ticket=ticket,
            category=classification.category,
            urgency=classification.urgency,
        )

# Training data
trainset = [
    dspy.Example(
        ticket="Can't log in to my account",
        category="account",
        urgency="medium",
        response="I can help you regain access. Please try resetting your password...",
    ).with_inputs("ticket"),
    dspy.Example(
        ticket="Production API returning 500 errors",
        category="technical",
        urgency="critical",
        response="I'm escalating this immediately. Our on-call team is investigating...",
    ).with_inputs("ticket"),
    # ... more examples
]

# Define a metric
def quality_metric(example, prediction, trace=None):
    return (
        prediction.category == example.category
        and prediction.urgency == example.urgency
    )

# Optimize
optimizer = dspy.BootstrapFewShot(metric=quality_metric, max_bootstrapped_demos=4)
optimized = optimizer.compile(SupportResponder(), trainset=trainset)

# Evaluate before saving
evaluator = Evaluate(devset=trainset[:10], metric=quality_metric, num_threads=4)
score = evaluator(optimized)
print(f"Score: {score}")

# Save the optimized state
optimized.save("support_responder_v1.json")
print("Saved optimized program to support_responder_v1.json")

# ==============================
# Part B: Load in production (run on every request)
# ==============================

import dspy

# Must configure LM before loading
lm = dspy.LM("openai/gpt-4o-mini", temperature=0.0)  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

# Create a fresh instance and load saved state
program = SupportResponder()
program.load("support_responder_v1.json")

# The program now has the optimized few-shot demos and instructions
result = program(ticket="I was charged twice on my last invoice")
print(f"Category: {result.category}")
print(f"Urgency: {result.urgency}")
print(f"Response: {result.response}")

# ==============================
# Part C: Version management
# ==============================

# Save new versions as you iterate
# optimized_v2.save("support_responder_v2.json")

# Compare versions
# program_v1 = SupportResponder()
# program_v1.load("support_responder_v1.json")
# score_v1 = evaluator(program_v1)
#
# program_v2 = SupportResponder()
# program_v2.load("support_responder_v2.json")
# score_v2 = evaluator(program_v2)
#
# print(f"v1: {score_v1}, v2: {score_v2}")

What to notice

• The class definition (SupportResponder) must exist in both the optimization script and the production code. load() restores learned state (demos, instructions) into the module's Predict sub-modules, but the Python logic in forward() comes from your code.
• Always call dspy.configure(lm=lm) before calling load(). The saved state does not include the LM configuration.
• You can switch models between optimization and production. For example, optimize with gpt-4o for better demos, then serve with gpt-4o-mini for lower costs. The few-shot demos still help the cheaper model.
• Version your saved files (e.g., _v1.json, _v2.json) and compare scores with Evaluate before deploying a new version.

DSPy Utilities API Reference

Condensed from dspy.ai. Verify against upstream for latest.

streamify

from dspy.streaming import streamify

streaming_program = streamify(
    program,                                        # Module -- required
    stream_listeners=None,                          # list[StreamListener]
    include_final_prediction_in_output_stream=True, # include Prediction in stream
    is_async_program=False,                         # True if program is already async
    async_streaming=True,                           # True for async generator
    status_message_provider=None,                   # custom status messages
)

StreamListener

from dspy.streaming import StreamListener

listener = StreamListener(
    signature_field_name,    # str -- required, output field to stream
    predict=None,            # predictor to monitor (auto-detected)
    predict_name=None,       # name identifier for the predictor
    allow_reuse=False,       # allow reuse across multiple streams
)

inspect_history

dspy.inspect_history(n=1)  # show last n LM calls

Shows full prompt sent, raw response, and adapter format.

save / load

program.save("path.json")   # save learned state (demos, instructions)
program.load("path.json")   # load into a fresh instance of the same class

Saves only learned state -- class definition must exist at load time. Call dspy.configure() before load().

asyncify

async_program = dspy.asyncify(program)
result = await async_program(**inputs)

Wraps sync DSPy programs for async execution. Captures and propagates dspy.configure settings to worker thread.

configure_cache

dspy.configure_cache(enable=True)   # enable/disable caching globally
dspy.LM("model", cache=False)      # per-LM cache control
dspy.configure_cache(restrict_pickle=True)  # 3.2+ - harden the on-disk cache against untrusted pickle payloads