Dspy Parallel

Run LM Calls in Parallel with dspy.Parallel

Guide the user through using DSPy's Parallel module to execute multiple LM calls concurrently. dspy.Parallel is the built-in way to speed up batch processing and fan-out patterns without writing threading code yourself.

What is dspy.Parallel

dspy.Parallel takes a list of (module, inputs) pairs and executes them concurrently using a thread pool. It handles threading, progress bars, error limits, and timeouts so you don't have to.

Use it when you have:

• A batch of inputs to run through the same module (classify 500 tickets, summarize 100 articles)
• Multiple independent modules to run on the same input (sentiment + topics + entities at once)
• Any set of LM calls that don't depend on each other

If call B depends on the result of call A, those two calls must be sequential. Everything else can be parallel.

Basic usage

Pass a list of (module, inputs) pairs. Each pair is one unit of work:

import dspy

lm = dspy.LM("openai/gpt-4o-mini")  # or any LiteLLM-supported provider
dspy.configure(lm=lm)

# A module to run on every input
classify = dspy.Predict("text -> label: str")

# A batch of inputs
texts = [
    "I love this product!",
    "Terrible experience, want a refund.",
    "It's okay, nothing special.",
    "Best purchase I've made this year.",
]

# Build execution pairs: (module, inputs_dict)
exec_pairs = [(classify, {"text": t}) for t in texts]

# Run them all in parallel
parallel = dspy.Parallel(num_threads=4)
results = parallel(exec_pairs)

for text, result in zip(texts, results):
    print(f"{text[:30]:30s} -> {result.label}")

results is a list in the same order as exec_pairs, so results[i] corresponds to exec_pairs[i].

Constructor options

dspy.Parallel(
    num_threads=4,               # number of concurrent threads (default: settings.num_threads)
    max_errors=5,                # stop after this many failures (default: settings.max_errors)
    return_failed_examples=False,# if True, return failures separately instead of raising
    provide_traceback=False,     # include tracebacks in error output
    disable_progress_bar=False,  # suppress the tqdm progress bar
    timeout=120,                 # max seconds per task before timeout
)

Configuring concurrency

Start with a thread count that matches your rate limits, not your CPU cores. LM calls are I/O-bound (waiting on HTTP responses), so you can safely use many threads:

# Conservative -- good starting point
parallel = dspy.Parallel(num_threads=4)

# Aggressive -- if your provider allows high concurrency
parallel = dspy.Parallel(num_threads=16)

# Match your provider's rate limit
# e.g., 60 requests/min = ~1/sec, so 4-8 threads keeps the pipeline full
parallel = dspy.Parallel(num_threads=8)

If you hit rate-limit errors (HTTP 429), reduce num_threads or add retry logic in your LM configuration.

Input formats

Parallel accepts inputs as dictionaries, dspy.Example objects, or tuples:

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

# Dict inputs (most common)
pairs = [(module, {"question": "What is DSPy?"})]

# dspy.Example inputs
example = dspy.Example(question="What is DSPy?").with_inputs("question")
pairs = [(module, example)]

# Both work the same way
parallel = dspy.Parallel(num_threads=2)
results = parallel(pairs)

Aggregating results

Results come back as a list. Aggregate however your application needs:

import dspy

classify = dspy.Predict("text -> label: str, confidence: float")
texts = ["Great!", "Terrible.", "Meh.", "Amazing!", "Awful."]

parallel = dspy.Parallel(num_threads=4)
results = parallel([(classify, {"text": t}) for t in texts])

# Count labels
from collections import Counter
label_counts = Counter(r.label for r in results)
print(label_counts)  # Counter({'positive': 2, 'negative': 2, 'neutral': 1})

# Filter by confidence
high_confidence = [
    (text, r.label)
    for text, r in zip(texts, results)
    if r.confidence > 0.8
]

# Build a summary dict
output = [
    {"text": t, "label": r.label, "confidence": r.confidence}
    for t, r in zip(texts, results)
]

Error handling

By default, Parallel raises an exception after max_errors failures. To handle errors gracefully, use return_failed_examples=True:

parallel = dspy.Parallel(
    num_threads=4,
    max_errors=10,
    return_failed_examples=True,
    provide_traceback=True,
)

results, failed_examples, exceptions = parallel(exec_pairs)

When return_failed_examples=True, the return value is a 3-tuple:

• `results` -- list of successful predictions (same length as successes)
• `failed_examples` -- list of (module, inputs) pairs that failed
• `exceptions` -- list of exceptions corresponding to each failure

Handle failures after the batch completes:

results, failed, errors = parallel(exec_pairs)

print(f"Succeeded: {len(results)}, Failed: {len(failed)}")

# Retry failures with a fallback module
if failed:
    fallback = dspy.ChainOfThought("text -> label: str")
    retry_pairs = [(fallback, inputs) for _, inputs in failed]
    retry_results = parallel(retry_pairs)

Setting an error budget

Use max_errors to fail fast when too many calls are failing (e.g., provider outage):

# Stop the whole batch if more than 5 calls fail
parallel = dspy.Parallel(num_threads=4, max_errors=5)

try:
    results = parallel(exec_pairs)
except Exception as e:
    print(f"Batch aborted: {e}")

Timeouts

The timeout parameter sets a per-task time limit in seconds. Tasks that exceed this are terminated:

# Give each task up to 60 seconds
parallel = dspy.Parallel(num_threads=4, timeout=60)

Using different modules per item

Each pair can use a different module. This is useful for fan-out patterns where you run multiple analyses on the same input:

import dspy

sentiment = dspy.Predict("text -> sentiment: str")
topics = dspy.Predict("text -> topics: list[str]")
summary = dspy.ChainOfThought("text -> summary: str")

text = "DSPy is a framework for programming language models..."

# Fan out: three different modules, same input
exec_pairs = [
    (sentiment, {"text": text}),
    (topics, {"text": text}),
    (summary, {"text": text}),
]

parallel = dspy.Parallel(num_threads=3)
results = parallel(exec_pairs)

combined = {
    "sentiment": results[0].sentiment,
    "topics": results[1].topics,
    "summary": results[2].summary,
}

When to use Parallel vs a sequential loop

Sequential loop (before)

# Slow: each call waits for the previous one to finish
results = []
for text in texts:
    result = classify(text=text)
    results.append(result)

Parallel (after)

# Fast: all calls run concurrently
parallel = dspy.Parallel(num_threads=8)
results = parallel([(classify, {"text": t}) for t in texts])

For a batch of 100 items with ~1 second per LM call:

• Sequential: ~100 seconds
• Parallel (8 threads): ~13 seconds

Parallel inside a module

Wrap Parallel usage inside a dspy.Module for clean composition:

class BatchClassifier(dspy.Module):
    def __init__(self, num_threads=4):
        self.classify = dspy.Predict("text -> label: str, confidence: float")
        self.num_threads = num_threads

    def forward(self, texts: list[str]):
        parallel = dspy.Parallel(num_threads=self.num_threads)
        exec_pairs = [(self.classify, {"text": t}) for t in texts]
        results = parallel(exec_pairs)

        return dspy.Prediction(
            labels=[r.label for r in results],
            confidences=[r.confidence for r in results],
        )

# Usage
classifier = BatchClassifier(num_threads=8)
result = classifier(texts=["Great!", "Terrible.", "Meh."])
print(result.labels)  # ["positive", "negative", "neutral"]

This keeps the parallelism as an implementation detail. Callers don't need to know about threading -- they just pass a list and get a list back.

Gotchas

1. Claude writes a `for` loop instead of using `dspy.Parallel`. When asked to process a batch of inputs, Claude defaults to a sequential loop. For any batch of 5+ independent LM calls, use dspy.Parallel — it is dramatically faster because LM calls are I/O-bound. 2. Claude sets `num_threads` to match CPU cores. LM calls are network-bound (waiting on HTTP responses), not CPU-bound. Thread count should match your provider rate limit, not your CPU count. 8-16 threads is typical even on a 4-core machine. 3. Claude forgets that `return_failed_examples=True` changes the return type. Without it, parallel(pairs) returns a flat list. With it, it returns a 3-tuple (results, failed_examples, exceptions). Destructure accordingly or the code will break. 4. Claude nests Parallel inside Parallel without considering total concurrency. An inner Parallel(num_threads=3) inside an outer Parallel(num_threads=4) creates up to 12 concurrent LM calls. This can exceed provider rate limits. Calculate the total: outer_threads * inner_threads. 5. Claude uses `dspy.Parallel` for 1-2 items. The threading overhead is not worth it for fewer than ~5 items. Just call the module directly.

Cross-references

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

• Modules are the building blocks you pass to Parallel -- see /dspy-modules
• Multi-step pipelines that combine sequential and parallel stages -- see /ai-building-pipelines
• Evaluation uses its own threading via num_threads -- see /dspy-evaluate
• For worked examples (batch classification, multi-aspect analysis), see examples.md
• 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

Additional resources

• dspy.Parallel API docs
• For constructor signatures and method reference, see reference.md
• For worked examples (batch classification, multi-aspect analysis), see examples.md

dspy-parallel Examples

Example 1: Batch classification in parallel

Classify a batch of support tickets by urgency and category, all at once:

import dspy
from typing import Literal

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


class ClassifyTicket(dspy.Signature):
    """Classify a support ticket by urgency and category."""
    ticket: str = dspy.InputField(desc="Customer support ticket text")
    urgency: Literal["low", "medium", "high", "critical"] = dspy.OutputField()
    category: Literal["billing", "technical", "account", "feature_request", "other"] = dspy.OutputField()


# Simulated batch of support tickets
tickets = [
    "My account was charged twice for the same subscription. Please refund ASAP.",
    "Would be cool if you added dark mode to the dashboard.",
    "I can't log in. Password reset emails aren't arriving. I have a demo in 30 minutes!",
    "How do I export my data to CSV?",
    "Our production integration is down. API returns 500 errors on every request.",
    "Can you change the email address on my account?",
    "The mobile app crashes when I open the settings page.",
    "I'd like to upgrade from the free plan to the team plan.",
]

# Build execution pairs
classify = dspy.Predict(ClassifyTicket)
exec_pairs = [(classify, {"ticket": t}) for t in tickets]

# Run classification in parallel
parallel = dspy.Parallel(num_threads=4)
results = parallel(exec_pairs)

# Print results
print(f"{'Ticket':<70} {'Urgency':<10} {'Category'}")
print("-" * 100)
for ticket, result in zip(tickets, results):
    print(f"{ticket[:67]+'...' if len(ticket)>67 else ticket:<70} {result.urgency:<10} {result.category}")

# Route critical tickets
critical_tickets = [
    (ticket, result)
    for ticket, result in zip(tickets, results)
    if result.urgency == "critical"
]
print(f"\n{len(critical_tickets)} critical ticket(s) need immediate attention:")
for ticket, result in critical_tickets:
    print(f"  [{result.category}] {ticket[:80]}")

Expected output:

Ticket                                                                 Urgency    Category
----------------------------------------------------------------------------------------------------
My account was charged twice for the same subscription. Please refu... high       billing
Would be cool if you added dark mode to the dashboard.                 low        feature_request
I can't log in. Password reset emails aren't arriving. I have a de... critical   account
How do I export my data to CSV?                                        low        technical
Our production integration is down. API returns 500 errors on every... critical   technical
Can you change the email address on my account?                        low        account
The mobile app crashes when I open the settings page.                  medium     technical
I'd like to upgrade from the free plan to the team plan.               low        billing

2 critical ticket(s) need immediate attention:
  [account] I can't log in. Password reset emails aren't arriving. I have a demo in 30 minutes!
  [technical] Our production integration is down. API returns 500 errors on every request.

With error handling

For production workloads, handle failures gracefully:

parallel = dspy.Parallel(
    num_threads=4,
    max_errors=3,
    return_failed_examples=True,
    provide_traceback=True,
)

results, failed, errors = parallel(exec_pairs)

print(f"Classified: {len(results)}, Failed: {len(failed)}")

# Retry failures one at a time as a fallback
for (module, inputs), error in zip(failed, errors):
    print(f"Failed on: {inputs['ticket'][:50]}... Error: {error}")
    try:
        result = module(**inputs)
        results.append(result)
    except Exception:
        print("  Retry also failed, skipping.")

---

Example 2: Parallel multi-aspect analysis

Analyze a single piece of text from multiple angles simultaneously -- sentiment, topics, and named entities -- then merge the results:

import dspy
from typing import Literal

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


class SentimentAnalysis(dspy.Signature):
    """Analyze the overall sentiment of the text."""
    text: str = dspy.InputField()
    sentiment: Literal["positive", "negative", "neutral", "mixed"] = dspy.OutputField()
    confidence: float = dspy.OutputField(desc="Confidence score from 0.0 to 1.0")
    explanation: str = dspy.OutputField(desc="Brief explanation of the sentiment")


class TopicExtraction(dspy.Signature):
    """Extract the main topics discussed in the text."""
    text: str = dspy.InputField()
    topics: list[str] = dspy.OutputField(desc="List of main topics, max 5")
    primary_topic: str = dspy.OutputField(desc="The single most prominent topic")


class EntityExtraction(dspy.Signature):
    """Extract named entities from the text."""
    text: str = dspy.InputField()
    people: list[str] = dspy.OutputField(desc="People mentioned")
    organizations: list[str] = dspy.OutputField(desc="Organizations mentioned")
    locations: list[str] = dspy.OutputField(desc="Locations mentioned")


# Three specialized modules
sentiment_module = dspy.Predict(SentimentAnalysis)
topic_module = dspy.Predict(TopicExtraction)
entity_module = dspy.Predict(EntityExtraction)

# Text to analyze
text = """
Apple announced its latest Vision Pro headset at WWDC in Cupertino yesterday.
CEO Tim Cook demonstrated the device's mixed reality capabilities to a packed audience.
While analysts from Goldman Sachs praised the innovation, consumer reviews on social
media were mixed -- many cited the $3,499 price tag as a significant barrier. Samsung
and Meta are expected to respond with competing products by Q2 2025.
"""

# Fan out: run all three analyses in parallel on the same text
exec_pairs = [
    (sentiment_module, {"text": text}),
    (topic_module, {"text": text}),
    (entity_module, {"text": text}),
]

parallel = dspy.Parallel(num_threads=3)
results = parallel(exec_pairs)

sentiment_result = results[0]
topic_result = results[1]
entity_result = results[2]

# Merge into a single analysis report
analysis = {
    "sentiment": {
        "label": sentiment_result.sentiment,
        "confidence": sentiment_result.confidence,
        "explanation": sentiment_result.explanation,
    },
    "topics": {
        "all": topic_result.topics,
        "primary": topic_result.primary_topic,
    },
    "entities": {
        "people": entity_result.people,
        "organizations": entity_result.organizations,
        "locations": entity_result.locations,
    },
}

print("=== Multi-Aspect Analysis ===\n")
print(f"Sentiment: {analysis['sentiment']['label']} "
      f"(confidence: {analysis['sentiment']['confidence']:.0%})")
print(f"  {analysis['sentiment']['explanation']}\n")
print(f"Topics: {', '.join(analysis['topics']['all'])}")
print(f"  Primary: {analysis['topics']['primary']}\n")
print(f"People: {', '.join(analysis['entities']['people'])}")
print(f"Organizations: {', '.join(analysis['entities']['organizations'])}")
print(f"Locations: {', '.join(analysis['entities']['locations'])}")

Expected output:

=== Multi-Aspect Analysis ===

Sentiment: mixed (confidence: 85%)
  Analysts praised the innovation but consumers criticized the high price

Topics: mixed reality, Vision Pro, consumer pricing, competition, WWDC
  Primary: Vision Pro

People: Tim Cook
Organizations: Apple, Goldman Sachs, Samsung, Meta
Locations: Cupertino

Wrapping it in a reusable module

For cleaner code, wrap the fan-out pattern in a dspy.Module:

class MultiAspectAnalyzer(dspy.Module):
    def __init__(self, num_threads=3):
        self.sentiment = dspy.Predict(SentimentAnalysis)
        self.topics = dspy.Predict(TopicExtraction)
        self.entities = dspy.Predict(EntityExtraction)
        self.num_threads = num_threads

    def forward(self, text: str):
        parallel = dspy.Parallel(num_threads=self.num_threads)
        results = parallel([
            (self.sentiment, {"text": text}),
            (self.topics, {"text": text}),
            (self.entities, {"text": text}),
        ])

        return dspy.Prediction(
            sentiment=results[0].sentiment,
            confidence=results[0].confidence,
            topics=results[1].topics,
            primary_topic=results[1].primary_topic,
            people=results[2].people,
            organizations=results[2].organizations,
            locations=results[2].locations,
        )


# Clean single-call interface
analyzer = MultiAspectAnalyzer()
result = analyzer(text=text)
print(result.sentiment, result.topics, result.people)

Scaling to a batch of documents

Combine both patterns -- parallelize across documents, with each document getting multi-aspect analysis:

documents = [text_1, text_2, text_3, ...]  # many documents

analyzer = MultiAspectAnalyzer(num_threads=3)

# Outer parallel: process documents concurrently
# Each call to analyzer internally fans out 3 modules in parallel
outer_parallel = dspy.Parallel(num_threads=4)
exec_pairs = [(analyzer, {"text": doc}) for doc in documents]
all_results = outer_parallel(exec_pairs)

for doc, result in zip(documents, all_results):
    print(f"Doc: {doc[:50]}... -> {result.sentiment}, {result.primary_topic}")

Note: the inner Parallel (3 threads per document) and outer Parallel (4 documents at once) combine for up to 12 concurrent LM calls. Make sure your provider rate limits can handle this.

Condensed from dspy.ai/api/modules/Parallel/. Verify against upstream for latest.

dspy.Parallel — API Reference

Constructor

dspy.Parallel(
    num_threads: int | None = None,
    max_errors: int | None = None,
    access_examples: bool = True,
    return_failed_examples: bool = False,
    provide_traceback: bool | None = None,
    disable_progress_bar: bool = False,
    timeout: int = 120,
    straggler_limit: int = 3,
)

Methods

__call__(exec_pairs, num_threads=None)

Delegates to forward().

forward(exec_pairs, num_threads=None)

Executes module-input pairs in parallel using a thread pool.

results = parallel(exec_pairs)
# or with return_failed_examples=True:
results, failed, exceptions = parallel(exec_pairs)

Returns:

• When return_failed_examples=False (default): list[Any] — results in same order as exec_pairs.
• When return_failed_examples=True: tuple[list[Any], list[tuple], list[Exception]] — (results, failed_pairs, exceptions).

Supported input formats

Each pair in exec_pairs is (module, inputs) where inputs can be:

Internal attributes