cmpnd

DSPy observability and deployment SDK for cmpnd.

Automatically trace your DSPy programs with one line of code, and deploy them with one more.

Installation

uv add cmpnd      # or: pip install cmpnd

Quick Start

import cmpnd

# Configure with your API key
cmpnd.configure(api_key="ck_xxx", project="my-project")

# Enable automatic DSPy instrumentation
cmpnd.auto_instrument()

# Your DSPy code is now automatically traced!
import dspy

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

cot = dspy.ChainOfThought("question -> answer")
result = cot(question="What is DSPy?")
# Trace is automatically sent to cmpnd

# Correlate the result with its platform trace (e.g. to attach feedback later):
cmpnd.trace_id(result)  # "a1b2c3d4-..."  (None if not produced under instrumentation)

Configuration

Using environment variables

export CMPND_API_KEY="ck_your_api_key"
export CMPND_ENDPOINT="https://platform.cmpnd.ai"  # optional

import cmpnd

cmpnd.configure()  # Reads from environment
cmpnd.auto_instrument()

Configuration options

cmpnd.configure(
    api_key="ck_xxx",                # Required: API key
    endpoint="https://platform.cmpnd.ai", # Optional: Backend URL
    project="my-project",            # Optional: Project name
    batch_size=100,                  # Optional: Batch size for export
    flush_interval_seconds=5.0,      # Optional: Flush interval
    capture_inputs=True,             # Optional: Capture function inputs
    capture_outputs=True,            # Optional: Capture function outputs
    default_tags={"env": "prod"},    # Optional: Tags for all traces
)

Deployment

Deploy a DSPy module to the cmpnd platform for server-side execution:

import cmpnd
import dspy

cmpnd.configure(api_key="ck_xxx")

lm = dspy.LM("groq/llama-3.1-8b-instant", temperature=0.0, max_tokens=16384)
dspy.configure(lm=lm)

module = dspy.Predict("question -> answer")
deployed = cmpnd.deploy(module)

# The returned program is callable — run it server-side like the local module:
result = deployed(question="What is DSPy?")
print(result.answer)

The SDK packages the module's source code into a ZIP, uploads it to the platform, and polls until the server-side parser finishes. Returns a DeployedProgram you can call directly to execute server-side; str(deployed) is its deployment_id and deployed.deployment_id exposes it explicitly.

With a metric (for optimization)

def accuracy(example, pred, trace=None):
    return example.answer == pred.answer

deployment_id = cmpnd.deploy(module, metric=accuracy)

Options

cmpnd.deploy(
    module,              # Required: a DSPy module
    metric=None,         # Optional: metric callable for optimization support
    timeout=300,         # Optional: seconds to wait for parsing (default 300)
)

How it works

1. The SDK generates source code from the live module (LM config, signature, types)
2. POST /api/v1/deployments creates a deployment and returns a presigned S3 upload URL
3. The source ZIP is uploaded directly to S3
4. The server-side parser extracts the program and LM configuration
5. The SDK polls until status is ready, then returns the DeployedProgram

The deployed program's LM model string (e.g. groq/llama-3.1-8b-instant) and configuration (temperature, max_tokens) are extracted automatically from the source.

Optimization

Optimize a DSPy program through the platform. cmpnd.optimize() mirrors cmpnd.deploy(): it returns a DeployedProgram you call directly — it just also tells you the program's score.

import cmpnd
import dspy

cmpnd.configure(api_key="ck_xxx")
dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))

program = dspy.ChainOfThought("topic -> haiku")

# gepa needs a 5-arg feedback metric returning a dspy.Prediction(score=, feedback=)
def metric(example, pred, trace=None, pred_name=None, pred_trace=None):
    ok = example.season.lower() not in pred.haiku.lower()
    return dspy.Prediction(score=float(ok),
                           feedback=None if ok else "Don't name the season.")

best = cmpnd.optimize(
    program,
    optimizer="gepa",                # or "bootstrap_fewshot", "random_search"
    trainset=train,
    valset=val,
    metric=metric,
    reflection_model="groq/llama-3.3-70b-versatile",   # gepa only
)

print(best.score)                       # best score found
print(best(topic="late autumn").haiku)  # call it directly — same as deploy()

The metric's arity must match the optimizer (checked before any network call): gepa needs a 5-arg feedback metric, while bootstrap_fewshot / random_search need a <=3-arg scalar metric.

Reusing an existing deployment

Skip the initial deploy by passing the deployment_id of a ready deployment as the starting program:

best = cmpnd.optimize(program, optimizer="bootstrap_fewshot",
                      trainset=train, valset=val, metric=metric,
                      deployment_id=deployed.deployment_id)

Options

cmpnd.optimize(
    program,                 # Required: the DSPy module to optimize
    optimizer="gepa",        # Required: "gepa" | "bootstrap_fewshot" | "random_search"
    trainset=train,          # Required: training examples
    valset=val,              # Required: validation examples
    metric=metric,           # Required: the evaluation metric (arity-checked)
    config=None,             # Optional: optimizer-specific config (allow-listed server-side)
    deployment_id=None,      # Optional: reuse a ready deployment as the starting program
    reflection_model=None,   # Optional (gepa): litellm model for the reflection LM
    max_steps=10,            # Optional: maximum optimize steps
    patience=2,              # Optional: stop after N consecutive non-improving steps
    seed=0,                  # Optional: starting seed; advanced each step
)

On failure, optimize() raises cmpnd.OptimizeError (a 4xx initiating a step — e.g. 409 no-data / mismatch) or cmpnd.OptimizeFailed (a step that transitioned to failed).

From the CLI

cmpnd optimize gepa task.py --reflection-model groq/llama-3.3-70b-versatile

The snippet (a .py file or - for stdin) binds program, trainset, valset, and metric; the CLI prints the best deployment id.

Custom Spans

Add custom spans to trace non-DSPy code:

Using the decorator

from cmpnd import trace, SpanType

@trace(name="fetch_documents", span_type=SpanType.RETRIEVE)
def fetch_documents(query: str) -> list[str]:
    # Your retrieval logic
    return documents

Using the context manager

from cmpnd import start_span, SpanType

def run_pipeline(query: str):
    with start_span("vector_search", span_type=SpanType.RETRIEVE) as span:
        span.set_attribute("index", "my-faiss-index")
        docs = search(query)
        span.set_outputs({"doc_count": len(docs)})

    return generate(query, docs)

What Gets Traced

The SDK automatically captures:

Module Execution

• Module type (Predict, ChainOfThought, ReAct, etc.)
• Signature name and instructions
• Input/output field names
• Demo count

LM Calls

• Model name and provider
• Token usage (prompt, completion, total)
• Request/response content

Adapters

• Format and parse operations
• Input/output transformations

Tools

• Tool name and description
• Invocation inputs/outputs

Evaluations

• Evaluation scores
• Program being evaluated

Span Types

Available span types for categorization:

• SpanType.MODULE - Generic DSPy module
• SpanType.PREDICT - Predict module
• SpanType.CHAIN_OF_THOUGHT - ChainOfThought module
• SpanType.REACT - ReAct agent
• SpanType.RETRIEVE - Retrieval operations
• SpanType.LM_CALL - Language model calls
• SpanType.ADAPTER_FORMAT - Adapter formatting
• SpanType.ADAPTER_PARSE - Adapter parsing
• SpanType.TOOL - Tool invocations
• SpanType.EVALUATION - Evaluation runs

API Reference

cmpnd.configure()

Initialize the SDK with your API key and options.

cmpnd.auto_instrument()

Automatically register the callback with DSPy.

cmpnd.CmpndCallback

The callback class for manual registration:

import dspy
import cmpnd

cmpnd.configure(api_key="ck_xxx")
dspy.configure(callbacks=[cmpnd.CmpndCallback()])

cmpnd.deploy()

Deploy a DSPy module to the cmpnd platform. Returns a callable DeployedProgram.

cmpnd.optimize()

Optimize a DSPy program through the platform. Returns the best DeployedProgram (call it directly; .score is the best score found). Raises cmpnd.OptimizeError / cmpnd.OptimizeFailed on failure.

cmpnd.execute()

Execute a deployed program by deployment_id, returning a dspy.Prediction. Usually you just call a DeployedProgram instead, which dispatches here.

cmpnd.DeployedProgram

A deployed, callable program returned by cmpnd.deploy() and cmpnd.optimize(). Call it with the original module's input fields; str() and .deployment_id give its id, and .score is its score when it came from optimize() (else None).

cmpnd.trace()

Decorator for custom traced functions.

cmpnd.start_span()

Context manager for custom spans.

cmpnd.get_current_trace()

Get the current trace (if any).

cmpnd.get_current_span()

Get the current span (if any).

cmpnd.trace_id()

Return the platform trace id stamped on a prediction returned by an instrumented program, or None if it wasn't produced under instrumentation. Use it to correlate a prediction with its trace, e.g. to attach feedback via POST /api/v1/traces/{cmpnd.trace_id(pred)}/metadata.

cmpnd.flush_exporter()

Force-publish everything the background exporter has queued, keeping it running. Use this when a mid-run reader needs a just-produced trace to exist server-side before fetching it — unlike shutdown_exporter(), the exporter stays alive so later spans still publish. Returns True if the drain completed within the timeout, False if it timed out or no exporter is running — check it before fetching so you don't read a trace that hasn't published yet.

cmpnd.shutdown_exporter()

Gracefully shutdown the background exporter.

License

MIT