Dspy Async

Run DSPy Modules Asynchronously

Guide the user through running DSPy modules with async/await for non-blocking execution in web frameworks, concurrent processing, and high-throughput applications.

What is async in DSPy

Every DSPy module supports async execution via aforward() and acall(). These return awaitable coroutines instead of blocking the event loop, making DSPy compatible with async web frameworks (FastAPI, Starlette, aiohttp) and enabling concurrent LM calls with asyncio.gather().

When to use async

Step 1: Basic async execution

Every DSPy module has an async variant:

import asyncio
import dspy

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

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

async def ask(question: str):
    # aforward() is the async version of forward()
    result = await qa.aforward(question=question)
    return result.answer

# Run it
answer = asyncio.run(ask("What is DSPy?"))
print(answer)

Two async methods:

• module.aforward(**kwargs) -- async version of module.forward()
• module.acall(**kwargs) -- async version of module(**kwargs) (same thing, convenience alias)

Step 2: Concurrent calls with asyncio.gather

Run multiple independent LM calls concurrently:

import asyncio
import dspy

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

summarizer = dspy.ChainOfThought("text -> summary")

async def summarize_batch(texts: list[str]):
    # Launch all summarizations concurrently
    tasks = [
        summarizer.aforward(text=text)
        for text in texts
    ]
    results = await asyncio.gather(*tasks)
    return [r.summary for r in results]

texts = ["Article 1...", "Article 2...", "Article 3..."]
summaries = asyncio.run(summarize_batch(texts))

This is significantly faster than sequential processing because LM calls are I/O-bound -- the network round-trip dominates.

Step 3: FastAPI endpoint

from fastapi import FastAPI
import dspy

app = FastAPI()

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

classifier = dspy.Predict("text -> label, confidence: float")

@app.post("/classify")
async def classify(text: str):
    # Non-blocking -- does not hold up other requests
    result = await classifier.aforward(text=text)
    return {"label": result.label, "confidence": result.confidence}

Why this matters: Without async, each request blocks the FastAPI worker thread. With aforward(), the worker is free to handle other requests while waiting for the LM response.

Step 4: Semaphore-based concurrency limiting

Prevent overwhelming the LM provider with too many concurrent requests:

import asyncio
import dspy

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

processor = dspy.ChainOfThought("input -> output")

# Limit to 10 concurrent LM calls
semaphore = asyncio.Semaphore(10)

async def process_one(input_text: str):
    async with semaphore:
        return await processor.aforward(input=input_text)

async def process_batch(inputs: list[str]):
    tasks = [process_one(text) for text in inputs]
    return await asyncio.gather(*tasks)

# Even with 1000 inputs, only 10 run concurrently
results = asyncio.run(process_batch(["input"] * 1000))

Step 5: Async with streaming

Combine async execution with streaming output:

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

app = FastAPI()

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

qa = dspy.ChainOfThought("question -> answer")
listener = StreamListener(signature_field_name="answer")
streaming_qa = streamify(qa, stream_listeners=[listener])

@app.get("/ask")
async def ask(question: str):
    async def generate():
        async for chunk in streaming_qa(question=question):
            if hasattr(chunk, "answer"):
                yield f"data: {chunk.answer}\n\n"
        yield "data: [DONE]\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")

Step 6: Async custom modules

When writing custom modules, implement aforward for async:

import dspy

class AsyncPipeline(dspy.Module):
    def __init__(self):
        self.classify = dspy.Predict("text -> category")
        self.summarize = dspy.ChainOfThought("text, category -> summary")

    async def aforward(self, text):
        # Run classification (async)
        classification = await self.classify.aforward(text=text)

        # Run summarization with the category (async)
        result = await self.summarize.aforward(
            text=text,
            category=classification.category,
        )
        return dspy.Prediction(
            category=classification.category,
            summary=result.summary,
        )

# Usage
pipeline = AsyncPipeline()
result = asyncio.run(pipeline.aforward(text="..."))

Step 7: Async with ReAct agents

Agents with MCP tools or async tool functions need acall():

import dspy

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

async def async_search(query: str) -> str:
    """Search the web asynchronously."""
    # Your async search implementation
    return "results..."

agent = dspy.ReAct("question -> answer", tools=[async_search])

async def run_agent(question: str):
    # acall() handles async tools automatically
    result = await agent.acall(question=question)
    return result.answer

Gotchas

1. Claude uses `module()` inside async functions instead of `await module.aforward()`. Calling a module synchronously inside an async function blocks the event loop. Always use aforward() or acall() in async contexts. 2. Claude nests `asyncio.run()` inside an existing event loop. You cannot call asyncio.run() from inside an async function -- it raises RuntimeError: This event loop is already running. Use await directly instead. 3. Claude forgets the semaphore for batch processing. Without a concurrency limit, asyncio.gather() with 1000 tasks hits rate limits immediately. Always add a semaphore when processing large batches. 4. Claude defines `forward()` but not `aforward()` in custom modules. If your module will be called with await, implement aforward(). DSPy does not auto-wrap forward() into an async version. 5. Claude mixes sync and async in the same pipeline. If one step is async (e.g., MCP tools), the entire call chain must be async. You cannot await inside a sync forward().

Additional resources

• dspy.ai/api/modules (aforward documentation)
• 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>

• Serving APIs with FastAPI -- see /ai-serving-apis
• Concurrent batch processing -- see /dspy-parallel
• Streaming output with async generators -- see /dspy-streaming
• MCP tools that require async -- see /dspy-mcp
• General utilities (caching, debugging) -- see /dspy-utils
• 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

Async DSPy Examples

Example 1: FastAPI with concurrent classification

A production endpoint that classifies multiple items concurrently:

from fastapi import FastAPI
from pydantic import BaseModel
import asyncio
import dspy

app = FastAPI()

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

classifier = dspy.Predict("text -> label, confidence: float")
semaphore = asyncio.Semaphore(20)  # Max 20 concurrent LM calls


class BatchRequest(BaseModel):
    texts: list[str]


class ClassificationResult(BaseModel):
    label: str
    confidence: float


@app.post("/classify-batch")
async def classify_batch(request: BatchRequest):
    async def classify_one(text: str):
        async with semaphore:
            result = await classifier.aforward(text=text)
            return ClassificationResult(
                label=result.label,
                confidence=float(result.confidence),
            )

    tasks = [classify_one(text) for text in request.texts]
    results = await asyncio.gather(*tasks)
    return {"results": results}

Example 2: Parallel research with timeout

Run multiple research queries concurrently with a timeout:

import asyncio
import dspy

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

researcher = dspy.ChainOfThought("topic -> findings, sources")


async def research_with_timeout(topics: list[str], timeout_seconds: float = 30.0):
    """Research multiple topics concurrently with a global timeout."""

    async def research_one(topic: str):
        result = await researcher.aforward(topic=topic)
        return {"topic": topic, "findings": result.findings}

    tasks = [research_one(topic) for topic in topics]

    try:
        results = await asyncio.wait_for(
            asyncio.gather(*tasks, return_exceptions=True),
            timeout=timeout_seconds,
        )
        # Filter out exceptions
        return [r for r in results if isinstance(r, dict)]
    except asyncio.TimeoutError:
        return [{"error": "Research timed out"}]


# Usage
topics = ["quantum computing advances", "CRISPR applications", "fusion energy progress"]
findings = asyncio.run(research_with_timeout(topics, timeout_seconds=20.0))
for f in findings:
    print(f"{f['topic']}: {f['findings'][:100]}...")

Example 3: Async pipeline with error recovery

A multi-step async pipeline that handles failures gracefully:

import asyncio
import dspy

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


class RobustPipeline(dspy.Module):
    def __init__(self):
        self.extract = dspy.Predict("document -> entities: list[str]")
        self.enrich = dspy.ChainOfThought("entity -> description, category")

    async def aforward(self, document):
        # Step 1: Extract entities
        extraction = await self.extract.aforward(document=document)
        entities = extraction.entities

        # Step 2: Enrich each entity concurrently (with error handling)
        async def enrich_one(entity: str):
            try:
                result = await self.enrich.aforward(entity=entity)
                return {
                    "entity": entity,
                    "description": result.description,
                    "category": result.category,
                }
            except Exception as e:
                return {"entity": entity, "error": str(e)}

        enriched = await asyncio.gather(*[enrich_one(e) for e in entities])

        return dspy.Prediction(
            entities=entities,
            enriched=[e for e in enriched if "error" not in e],
            errors=[e for e in enriched if "error" in e],
        )


pipeline = RobustPipeline()
result = asyncio.run(pipeline.aforward(document="Apple announced new AI features..."))
print(f"Enriched {len(result.enriched)} entities, {len(result.errors)} errors")

Async DSPy API Reference

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

aforward()

Available on all DSPy modules (Predict, ChainOfThought, ReAct, CodeAct, custom Module subclasses).

result = await module.aforward(**kwargs)

Returns: dspy.Prediction (same as sync version)

acall()

Convenience alias for aforward(). Identical behavior:

# These are equivalent
result = await module.acall(question="...")
result = await module.aforward(question="...")

Async patterns

Basic async call

import asyncio
import dspy

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

async def main():
    result = await module.aforward(question="...")
    return result.answer

asyncio.run(main())

Concurrent calls (asyncio.gather)

async def concurrent_calls(inputs: list[str]):
    tasks = [module.aforward(question=q) for q in inputs]
    return await asyncio.gather(*tasks)

Semaphore-limited concurrency

semaphore = asyncio.Semaphore(10)

async def limited_call(**kwargs):
    async with semaphore:
        return await module.aforward(**kwargs)

Timeout

async def with_timeout(**kwargs):
    return await asyncio.wait_for(
        module.aforward(**kwargs),
        timeout=30.0,
    )

Custom async modules

class MyModule(dspy.Module):
    def __init__(self):
        self.step1 = dspy.Predict("input -> intermediate")
        self.step2 = dspy.ChainOfThought("intermediate -> output")

    async def aforward(self, input):
        mid = await self.step1.aforward(input=input)
        result = await self.step2.aforward(intermediate=mid.intermediate)
        return result

Batch processing

DSPy modules also have a .batch() method for built-in batch processing:

# Sync batch
results = module.batch(
    [{"question": q} for q in questions],
    num_threads=10,
)

# For async batch processing, use asyncio.gather with semaphore

Framework integration

FastAPI

@app.post("/endpoint")
async def endpoint(input: str):
    result = await module.aforward(input=input)
    return {"output": result.output}

Starlette

async def homepage(request):
    result = await module.aforward(input=request.query_params["q"])
    return JSONResponse({"output": result.output})