# Pattern Multi Agent

> Multi-agent DSPy patterns and coordination

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

---


# Multi-Agent DSPy

## 🎯 Trigger Conditions
Use when asked about multi-agent systems, agent coordination, or distributed DSPy execution.

## 📚 Prerequisites
- `dspy` package installed
- Understanding of multi-agent architectures
- Coordination mechanism defined

## 🛠️ Multi-Agent Patterns

### 1. Agent Specialization
```python
class SpecialistAgent(dspy.Module):
    def __init__(self, specialty):
        self.specialty = specialty
        self.predictor = dspy.Predict(f"input -> {specialty}_output")
    
    def forward(self, input):
        return self.predictor(input=input)

# Create specialized agents
agents = {
    "math": SpecialistAgent("math"),
    "science": SpecialistAgent("science"),
    "general": SpecialistAgent("general")
}

# Route to specialist
def route_to_specialist(input):
    intent = classify_intent(input)
    return agents[intent](input)
```

### 2. Agent Debate
```python
class AgentDebate(dspy.Module):
    def __init__(self, n_agents=3):
        self.agents = [
            dspy.Predict("input -> opinion") for _ in range(n_agents)
        ]
        self.judge = dspy.Predict("opinions -> decision")
    
    def forward(self, input):
        # Generate opinions
        opinions = [agent(input=input).opinion for agent in self.agents]
        
        # Judge the debate
        decision = self.judge(opinions=opinions).decision
        return decision
```

### 3. Agent Hierarchy
```python
class HierarchicalAgent(dspy.Module):
    def __init__(self):
        self.manager = dspy.Predict("task -> subtasks")
        self.workers = [
            dspy.Predict("subtask -> result") for _ in range(4)
        ]
        self.aggregator = dspy.Predict("results -> final_answer")
    
    def forward(self, task):
        # Manager breaks down task
        subtasks = self.manager(task=task).subtasks
        
        # Workers execute subtasks
        results = [worker(subtask=subtask).result for worker, subtask in zip(self.workers, subtasks)]
        
        # Aggregate results
        return self.aggregator(results=results).final_answer
```

### 4. Agent Marketplace
```python
class AgentMarketplace(dspy.Module):
    def __init__(self):
        self.bids = {}
        self.agents = {}
    
    def register_agent(self, agent, cost):
        self.agents[agent] = cost
    
    def bid(self, agent, task):
        self.bids[task] = self.agents[agent]
    
    def execute(self, task):
        # Find lowest cost agent
        cheapest_agent = min(self.bids.items(), key=lambda x: x[1])
        return cheapest_agent[0](task)
```

## ⚠️ Pitfalls
- **Coordination overhead**: Communication between agents adds latency
- **Consistency**: Ensuring consistent behavior across agents
- **Resource usage**: Multiple agents consume more resources
- **Debugging**: Multi-agent systems are harder to debug

## 📖 References
- [Multi-Agent Systems](https://arxiv.org/abs/2308.10848)
- [DSPy Multi-Agent](https://dspy-docs.vercel.app/docs/patterns/multi-agent)