dspy-production-deployment

DSPy Production Deployment

Goal

Prepare a DSPy program for repeatable, observable, scalable, and safer production execution.

Cache Hardening

DSPy enables memory and disk caches by default. Disk cache deserialization uses pickle unless restricted. Enable the allowlist mode in production:

import dspy

dspy.configure_cache(restrict_pickle=True)

Register trusted custom cache types only when needed:

dspy.configure_cache(
    restrict_pickle=True,
    safe_types=[MyResult, Metadata],
)

Disable a cache layer explicitly when a deployment cannot persist data or requires fresh model responses:

dspy.configure_cache(
    enable_disk_cache=False,
    enable_memory_cache=True,
)

Save and Load

Prefer state-only JSON for readable, safer artifacts:

compiled.save("./artifacts/program.json", save_program=False)

loaded = MyProgram()
loaded.load("./artifacts/program.json")

Use whole-program save only for trusted artifacts. It uses cloudpickle:

compiled.save("./artifacts/program/", save_program=True)
loaded = dspy.load("./artifacts/program/")

Keep the DSPy major version compatible when loading saved programs.

Usage Tracking

dspy.configure(
    lm=dspy.LM("openai/gpt-4o-mini"),
    track_usage=True,
)

prediction = program(question="What is DSPy?")
print(prediction.get_lm_usage())

Cached calls return no new token usage.

Async Execution

Most built-in modules support acall():

import asyncio

async def main():
    prediction = await program.acall(question="What is DSPy?")
    print(prediction.answer)

asyncio.run(main())

Implement aforward() for custom async modules. Use dspy.asyncify(program) only when adapting a synchronous callable is the right boundary.

Streaming

import asyncio
import dspy

stream_program = dspy.streamify(
    dspy.Predict("question -> answer"),
    stream_listeners=[
        dspy.streaming.StreamListener(signature_field_name="answer"),
    ],
)

async def main():
    async for chunk in stream_program(question="Explain DSPy briefly."):
        print(chunk)

asyncio.run(main())