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

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

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

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

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
• DSPy Pipelines