# Pipeline Async

> Async execution patterns for DSPy pipelines

- Skill: `j33bs/pipeline-async` (Agent Skill)
- Install (CLI): `npx skillmds@latest add j33bs/pipeline-async`
- Raw SKILL.md: https://api.skillmd.com/api/skills/j33bs/pipeline-async/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: DevOps & Infra
- Author: j33bs (https://skillmd.com/u/j33bs)
- Updated: 2026-09-22
- Page: https://skillmd.com/skills/j33bs/pipeline-async

---


# Pipeline Async Execution

## 🎯 Trigger Conditions
Use when asked about asynchronous DSPy execution, parallel processing, or non-blocking pipelines.

## 📚 Prerequisites
- `dspy` package installed
- Python 3.10+ (for async/await)

## 🛠️ Async Patterns

### 1. Basic Async Execution
```python
import asyncio
import dspy

async def async_predict(input_data):
    predictor = dspy.Predict("input -> output")
    result = await asyncio.to_thread(predictor, input=input_data)
    return result.output

# Run multiple predictions concurrently
async def batch_predict(inputs):
    tasks = [async_predict(inp) for inp in inputs]
    results = await asyncio.gather(*tasks)
    return results
```

### 2. Async Pipeline
```python
class AsyncPipeline(dspy.Module):
    def __init__(self):
        self.step1 = dspy.Predict("input -> intermediate")
        self.step2 = dspy.Predict("intermediate -> output")
    
    async def forward(self, input):
        # Run steps asynchronously
        intermediate = await asyncio.to_thread(self.step1, input=input)
        output = await asyncio.to_thread(self.step2, intermediate=intermediate.intermediate)
        return output
```

### 3. Async with Tool Use
```python
async def async_tool_use(query):
    # Simulate async tool call
    await asyncio.sleep(1)  # Simulate network delay
    return "Tool result"

async def async_react(query):
    # Run multiple tool calls concurrently
    tools = [async_tool_use(f"{query}_{i}") for i in range(3)]
    results = await asyncio.gather(*tools)
    return results
```

### 4. Async Batch Processing
```python
async def async_batch_process(items, batch_size=10):
    results = []
    for i in range(0, len(items), batch_size):
        batch = items[i:i+batch_size]
        batch_results = await asyncio.gather(*[
            asyncio.to_thread(process_item, item)
            for item in batch
        ])
        results.extend(batch_results)
    return results
```

## ⚠️ Pitfalls
- **Resource limits**: Async doesn't reduce total resource usage
- **Error handling**: Async errors require special handling
- **Blocking calls**: Avoid synchronous calls in async functions
- **Debugging**: Async can be harder to debug

## 📖 References
- [Python Asyncio](https://docs.python.org/3/library/asyncio.html)
- [DSPy Pipelines](https://dspy-docs.vercel.app/docs/deep-dive/pipelines)