ml-dspy-multi-agent

name	ml-dspy-multi-agent
description	Multi-agent systems with DSPy including orchestration, GEPA optimization, and inter-agent communication

DSPy Multi-Agent Systems

Scope: Multi-agent architectures, orchestration, GEPA, agent communication, coordination Lines: ~490 Last Updated: 2025-10-30

When to Use This Skill

Activate this skill when:

• Building systems with multiple specialized agents
• Implementing hierarchical agent architectures (manager-worker)
• Creating collaborative agent networks
• Optimizing multi-agent systems with GEPA
• Designing domain-specific agent teams (research, customer service, etc.)
• Implementing agent communication and coordination protocols

Core Concepts

Multi-Agent Systems

Definition: Multiple autonomous agents working together to solve complex tasks

Purpose:

• Specialization: Each agent focuses on specific domain/skill
• Scalability: Distribute workload across agents
• Robustness: System continues if one agent fails
• Modularity: Easy to add/remove/update agents

Key insight: Divide complex problems among specialized agents with clear roles

Agent Architectures

Hierarchical: Manager agent coordinates worker agents

• Manager: Plans, delegates, synthesizes
• Workers: Execute specialized tasks
• Clear command structure

Peer-to-Peer: Agents collaborate as equals

• Distributed decision making
• Consensus-based coordination
• No single point of failure

Pipeline: Sequential agent chain

• Each agent processes and passes to next
• Clear data flow
• Easy to debug

Network: Agents communicate freely

• Dynamic collaboration
• Complex coordination
• Maximum flexibility

GEPA Optimization

GEPA: General-to-specific Evolutionary Prompt Augmentation

Purpose: Optimize multi-agent systems jointly

• Co-evolves agent prompts
• Considers inter-agent dependencies
• Improves system-wide performance

When to use:

• Multiple agents with interdependencies
• Need to optimize entire system (not just individual agents)
• Complex multi-step workflows

---

Patterns

Pattern 1: Hierarchical Multi-Agent System

import dspy

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

# Define worker agents
class ResearchAgent(dspy.Module):
    """Research agent specialized in information gathering."""

    def __init__(self):
        super().__init__()
        self.search = dspy.Retrieve(k=5)
        self.synthesize = dspy.ChainOfThought("topic, sources -> summary")

    def forward(self, topic):
        sources = self.search(topic).passages
        return self.synthesize(topic=topic, sources="\n".join(sources))

class AnalysisAgent(dspy.Module):
    """Analysis agent specialized in data interpretation."""

    def __init__(self):
        super().__init__()
        self.analyze = dspy.ChainOfThought("data, question -> analysis, insights: list[str]")

    def forward(self, data, question):
        return self.analyze(data=data, question=question)

class WritingAgent(dspy.Module):
    """Writing agent specialized in content creation."""

    def __init__(self):
        super().__init__()
        .write = dspy.ChainOfThought()

     ():
         .write(topic=topic, content=content, style=style)


 (dspy.Module):
    

     ():
        ().__init__()

        
        .researcher = ResearchAgent()
        .analyst = AnalysisAgent()
        .writer = WritingAgent()

        
        .planner = dspy.ChainOfThought()
        .synthesizer = dspy.ChainOfThought()

     ():
        
        plan = .planner(task=task)

        
         (plan.plan, ):
            steps = [s.strip()  s  plan.plan.split()]
        :
            steps = plan.plan

        
        results = []

         step  steps[:]:  
            step_lower = step.lower()

               step_lower    step_lower:
                result = .researcher(topic=step)
                results.append()

               step_lower    step_lower:
                
                data = .join(results)  results  
                result = .analyst(data=data, question=step)
                results.append()

               step_lower    step_lower:
                content = .join(results)  results  
                result = .writer(topic=task, content=content)
                results.append()

            :
                
                result = .researcher(topic=step)
                results.append()

        
        all_results = .join(results)
        final = .synthesizer(task=task, results=all_results)

         dspy.Prediction(
            answer=final.final_answer,
            steps=steps,
            results=results
        )


manager = ManagerAgent()
result = manager(task=)

(result.answer)
()

Benefits:

• Clear separation of concerns
• Specialized agents for different tasks
• Centralized coordination
• Easy to add new worker agents

Pattern 2: Peer-to-Peer Agent Collaboration

import dspy

class CollaborativeAgent(dspy.Module):
    """Agent that can consult peers."""

    def __init__(self, name, specialty, peers=None):
        super().__init__()
        self.name = name
        self.specialty = specialty
        self.peers = peers or []

        self.decide = dspy.ChainOfThought(
            "task, specialty -> can_handle: bool, needs_peer: bool, peer_name"
        )
        self.execute = dspy.ChainOfThought(f"task, context -> result")

    def add_peer(self, peer):
        """Add peer agent."""
        if peer not in self.peers:
            self.peers.append(peer)

    def forward(self, task, context="", visited=None):
        if visited is None:
            visited = set()

        # Prevent infinite loops
        if self.name in visited:
            return dspy.Prediction(result="Already consulted this agent")

        visited.add(self.name)

        # Decide if this agent can handle task
        decision = self.decide(task=task, specialty=.specialty)

        can_handle = (decision.can_handle).lower()  [, , ]
        needs_peer = (decision.needs_peer).lower()  [, , ]

        
         can_handle:
            result = .execute(task=task, context=context)
            agent_result = 

            
             needs_peer  .peers:
                peer_name = decision.peer_name
                peer = ((p  p  .peers  p.name == peer_name), )

                 peer:
                    peer_result = peer(task=task, context=agent_result, visited=visited)
                    combined = 
                     dspy.Prediction(result=combined)

             dspy.Prediction(result=agent_result)

        
         needs_peer  .peers:
            peer_name = decision.peer_name
            peer = ((p  p  .peers  p.name == peer_name), )

             peer:
                 peer(task=task, context=context, visited=visited)

         dspy.Prediction(result=)


search_agent = CollaborativeAgent(, )
code_agent = CollaborativeAgent(, )
writing_agent = CollaborativeAgent(, )


search_agent.add_peer(code_agent)
search_agent.add_peer(writing_agent)
code_agent.add_peer(search_agent)
code_agent.add_peer(writing_agent)
writing_agent.add_peer(search_agent)
writing_agent.add_peer(code_agent)


result = search_agent(task=)
(result.result)

Benefits:

• No single point of failure
• Agents can route tasks dynamically
• Flexible collaboration
• Emergent behavior

Pattern 3: Sequential Pipeline

import dspy

class PipelineStage(dspy.Module):
    """Base class for pipeline stages."""

    def __init__(self, name):
        super().__init__()
        self.name = name

    def forward(self, input_data):
        raise NotImplementedError

class ExtractionStage(PipelineStage):
    """Extract structured data."""

    def __init__(self):
        super().__init__("Extraction")
        self.extract = dspy.ChainOfThought(
            "text -> entities: list[str], facts: list[str]"
        )

    def forward(self, input_data):
        result = self.extract(text=input_data)
        return {
            'entities': result.entities,
            'facts': result.facts,
            'source': input_data
        }

class EnrichmentStage(PipelineStage):
    """Enrich data with additional context."""

    def __init__(self):
        super().__init__("Enrichment")
        self.retrieve = dspy.Retrieve(k=3)
        self.enrich = dspy.ChainOfThought()

     ():
        entities = input_data[]

        
        all_sources = []
         entity  entities[:]:  
            sources = .retrieve(entity).passages
            all_sources.extend(sources)

        
        enriched = .enrich(
            entities=.join(entities),
            sources=.join(all_sources[:])
        )

         {
            **input_data,
            : enriched.enriched_data
        }

 ():
    

     ():
        ().__init__()
        .synthesize = dspy.ChainOfThought(
            
        )

     ():
        result = .synthesize(
            facts=.join(input_data[]),
            enriched_data=input_data[]
        )

         {
            **input_data,
            : result.summary,
            : result.key_points
        }

 (dspy.Module):
    

     ():
        ().__init__()
        .stages = stages

     ():
        data = input_data

        
         stage  .stages:
            ()
            data = stage(data)

         dspy.Prediction(**data)


pipeline = MultiAgentPipeline(stages=[
    ExtractionStage(),
    EnrichmentStage(),
    SynthesisStage()
])


result = pipeline(input_data=)
()
()

Benefits:

• Clear data flow
• Easy to debug
• Modular stages
• Predictable execution

Pattern 4: Multi-Agent RAG with Specialization

import dspy

class SpecializedRAGAgent(dspy.Module):
    """RAG agent specialized for a domain."""

    def __init__(self, domain, collection_name):
        super().__init__()
        self.domain = domain
        self.retrieve = dspy.Retrieve(k=5)  # Would be domain-specific
        self.generate = dspy.ChainOfThought(f"context, question -> answer, confidence: float")

    def forward(self, question):
        # Retrieve from domain-specific collection
        passages = self.retrieve(question).passages
        context = "\n\n".join(passages)

        # Generate answer
        result = self.generate(context=context, question=question)

        try:
            conf = float(result.confidence)
        except:
            conf = 0.5

        return dspy.Prediction(
            answer=result.answer,
            confidence=conf,
            domain=self.domain
        )

class MultiDomainRAG(dspy.Module):
    """Multi-agent RAG system with domain routing."""

    def __init__(self):
        super().__init__()

        # Domain-specific agents
        self.agents = {
            'technical': SpecializedRAGAgent('technical', 'tech_docs'),
            : SpecializedRAGAgent(, ),
            : SpecializedRAGAgent(, ),
        }

        
        .router = dspy.Predict()

        
        .aggregate = dspy.ChainOfThought(
            
        )

     ():
        
        routing = .router(question=question)
        domain = routing.domain.lower()

        
         domain  .agents:
            primary = .agents[domain](question)

            :
                route_conf = (routing.confidence)
            :
                route_conf = 

            
             route_conf >   primary.confidence > :
                 primary

        
        answers = []
         domain_name, agent  .agents.items():
            result = agent(question)
            answers.append()

        
        all_answers = .join(answers)
        final = .aggregate(question=question, answers=all_answers)

         dspy.Prediction(
            answer=final.final_answer,
            sources=answers
        )


lm = dspy.LM()
dspy.configure(lm=lm)

rag = MultiDomainRAG()
result = rag(question=)
(result.answer)

Benefits:

• Domain-specific expertise
• Better retrieval quality
• Intelligent routing
• Fallback to multiple domains

Pattern 5: GEPA-Optimized Multi-Agent System

import dspy

# Define specialized agents
class Agent1(dspy.Module):
    """First agent in pipeline."""

    def __init__(self):
        super().__init__()
        self.process = dspy.ChainOfThought("input -> intermediate_output")

    def forward(self, input):
        return self.process(input=input)

class Agent2(dspy.Module):
    """Second agent that depends on Agent1."""

    def __init__(self):
        super().__init__()
        self.refine = dspy.ChainOfThought("input, previous_output -> refined_output")

    def forward(self, input, previous_output):
        return self.refine(input=input, previous_output=previous_output)

class Agent3(dspy.Module):
    """Final agent that synthesizes."""

    def __init__(self):
        super().__init__()
        self.synthesize = dspy.ChainOfThought("input, context -> final_answer")

    def ():
         .synthesize(=, context=context)


 (dspy.Module):
    

     ():
        ().__init__()
        .agent1 = Agent1()
        .agent2 = Agent2()
        .agent3 = Agent3()

     ():
        
        result1 = .agent1(=question)
        result2 = .agent2(=question, previous_output=result1.intermediate_output)
        result3 = .agent3(=question, context=result2.refined_output)

         dspy.Prediction(answer=result3.final_answer)


trainset = [
    dspy.Example(
        question=,
        answer=
    ).with_inputs(),
    
]


 ():
     example.answer.lower()  pred.answer.lower()



 dspy.teleprompt  GEPA

optimizer = GEPA(
    metric=accuracy,
    breadth=,  
    depth=,    
    init_temperature=
)


system = MultiAgentSystem()
optimized_system = optimizer.(
    student=system,
    trainset=trainset,
    max_bootstrapped_demos=,
)


result = optimized_system(question=)
(result.answer)

GEPA Benefits:

• Co-optimizes all agents jointly
• Considers inter-agent dependencies
• Better than optimizing agents independently
• Evolutionary approach to prompt generation

Pattern 6: Agent Communication Protocol

import dspy
from dataclasses import dataclass
from typing import Optional

@dataclass
class Message:
    """Message passed between agents."""
    sender: str
    recipient: str
    content: str
    message_type: str  # 'request', 'response', 'broadcast'
    context: Optional[dict] = None

class CommunicatingAgent(dspy.Module):
    """Agent with messaging capability."""

    def __init__(self, name, role):
        super().__init__()
        self.name = name
        self.role = role
        self.inbox = []

        self.process_message = dspy.ChainOfThought(
            "message, role -> response, action"
        )

    def send_message(self, recipient, content, message_type='request', context=None):
        """Send message to another agent."""
        return Message(
            sender=self.name,
            recipient=recipient,
            content=content,
            message_type=message_type,
            context=context
        )

    def receive_message(self, message: Message):
        """Receive message from another agent."""
        self.inbox.append(message)

     ():
        
        responses = []

         msg  .inbox:
            result = .process_message(
                message=msg.content,
                role=.role
            )

            responses.append(
                .send_message(
                    recipient=msg.sender,
                    content=result.response,
                    message_type=,
                    context={: result.action}
                )
            )

        .inbox = []  
         responses

     ():
        
        result = .process_message(message=task, role=.role)
         dspy.Prediction(
            response=result.response,
            action=result.action
        )

 :
    

     ():
        .agents = {}

     ():
        
        .agents[agent.name] = agent

     ():
        
         message.recipient == :
            
             name, agent  .agents.items():
                 name != message.sender:
                    agent.receive_message(message)
         message.recipient  .agents:
            .agents[message.recipient].receive_message(message)
        :
            ()

     ():
        
        all_responses = []

         agent  .agents.values():
            responses = agent.process_inbox()
            all_responses.extend(responses)

            
             response  responses:
                .deliver_message(response)

         all_responses


researcher = CommunicatingAgent(, )
analyst = CommunicatingAgent(, )
writer = CommunicatingAgent(, )


broker = MessageBroker()
broker.register_agent(researcher)
broker.register_agent(analyst)
broker.register_agent(writer)


msg = researcher.send_message(
    recipient=,
    content=,
    message_type=
)
broker.deliver_message(msg)


responses = broker.process_all()
()

Benefits:

• Structured communication
• Broadcast capability
• Message routing
• Clear message types

Pattern 7: Consensus-Based Multi-Agent

import dspy

class VotingAgent(dspy.Module):
    """Agent that can vote on proposals."""

    def __init__(self, name, expertise):
        super().__init__()
        self.name = name
        self.expertise = expertise
        self.vote = dspy.ChainOfThought(
            "proposal, expertise -> vote: bool, confidence: float, reasoning"
        )

    def forward(self, proposal):
        result = self.vote(proposal=proposal, expertise=self.expertise)

        vote_bool = str(result.vote).lower() in ['true', 'yes', '1']

        try:
            conf = float(result.confidence)
        except:
            conf = 0.5

        return dspy.Prediction(
            vote=vote_bool,
            confidence=conf,
            reasoning=result.reasoning,
            agent=self.name
        )

class ConsensusSystem(dspy.Module):
    """Multi-agent system using consensus voting."""

    def __init__(self, agents, threshold=0.6):
        super().__init__()
        self.agents = agents
        self.threshold = threshold

        self.proposer = dspy.ChainOfThought("question -> proposal")
        self.synthesizer = dspy.ChainOfThought(
            
        )

     ():
        
        proposal_result = .proposer(question=question)
        proposal = proposal_result.proposal

        
        votes = []
         agent  .agents:
            vote_result = agent(proposal)
            votes.append({
                : vote_result.agent,
                : vote_result.vote,
                : vote_result.confidence,
                : vote_result.reasoning
            })

        
        positive_votes = (  v  votes  v[])
        consensus_score = positive_votes / (votes)

        
        votes_summary = .join([
            
             v  votes
        ])

        final = .synthesizer(
            question=question,
            proposal=proposal,
            votes=votes_summary
        )

         dspy.Prediction(
            answer=final.final_answer,
            proposal=proposal,
            consensus_score=consensus_score,
            reached_consensus=consensus_score >= .threshold,
            votes=votes
        )


agents = [
    VotingAgent(, ),
    VotingAgent(, ),
    VotingAgent(, ),
]


consensus = ConsensusSystem(agents, threshold=)


result = consensus(question=)
()
()
()

Benefits:

• Democratic decision making
• Multiple perspectives
• Transparent reasoning
• Configurable thresholds

Pattern 8: Adaptive Multi-Agent System

import dspy

class AdaptiveMultiAgent(dspy.Module):
    """System that dynamically selects and coordinates agents."""

    def __init__(self, agent_pool):
        super().__init__()
        self.agent_pool = agent_pool  # Dict of {name: agent}

        self.selector = dspy.ChainOfThought(
            "task, available_agents -> selected_agents: list[str], strategy"
        )
        self.coordinator = dspy.ChainOfThought(
            "task, strategy, agent_results -> final_answer"
        )

    def forward(self, task):
        # Describe available agents
        agents_desc = ", ".join([
            f"{name}: {agent.__doc__ or 'No description'}"
            for name, agent in self.agent_pool.items()
        ])

        # Select agents for this task
        selection = self.selector(task=task, available_agents=agents_desc)

        # Parse selected agents
        if isinstance(selection.selected_agents, str):
            selected = [a.strip() for a in selection.selected_agents.split(',')]
        else:
            selected = selection.selected_agents

        # Execute selected agents
        results = []
        for agent_name in selected[:5]:  
             agent_name  .agent_pool:
                agent = .agent_pool[agent_name]
                :
                    result = agent(task)
                    results.append()
                 Exception  e:
                    results.append()

        
        all_results = .join(results)
        final = .coordinator(
            task=task,
            strategy=selection.strategy,
            agent_results=all_results
        )

         dspy.Prediction(
            answer=final.final_answer,
            agents_used=selected,
            strategy=selection.strategy
        )


agent_pool = {
    : dspy.Predict(),
    : dspy.ChainOfThought(),
    : dspy.ChainOfThought(),
    : dspy.Predict(),
}


adaptive = AdaptiveMultiAgent(agent_pool)


result = adaptive(task=)
()
()

Benefits:

• Dynamic agent selection
• Task-specific configuration
• Resource efficient
• Flexible architecture

---

Quick Reference

Multi-Agent Architectures

# Hierarchical
manager = ManagerAgent(workers=[agent1, agent2, agent3])

# Peer-to-peer
agent1.add_peer(agent2)
agent2.add_peer(agent1)

# Pipeline
pipeline = Sequential([stage1, stage2, stage3])

# Adaptive
adaptive = AdaptiveSystem(agent_pool={name: agent, ...})

GEPA Optimization

from dspy.teleprompt import GEPA

optimizer = GEPA(
    metric=metric_fn,
    breadth=5,
    depth=2,
)

optimized = optimizer.compile(
    student=multi_agent_system,
    trainset=trainset,
)

Best Practices

✅ DO: Specialize agents for distinct roles
✅ DO: Limit number of agents (5-10 max)
✅ DO: Define clear communication protocols
✅ DO: Handle agent failures gracefully
✅ DO: Optimize system jointly with GEPA
✅ DO: Log inter-agent communications

❌ DON'T: Create too many similar agents
❌ DON'T: Allow circular dependencies
❌ DON'T: Optimize agents independently (use GEPA)
❌ DON'T: Ignore agent failures
❌ DON'T: Forget to set max iterations

---

Anti-Patterns

❌ Too many agents: Coordination overhead

# Bad - 50 agents
system = MultiAgent(agents=list_of_50_agents)

✅ 5-10 focused agents:

# Good
system = MultiAgent(agents=[search, analyze, write, validate])

❌ Circular dependencies: Infinite loops

# Bad
agent1 → agent2 → agent3 → agent1  # Loop!

✅ Acyclic flow or loop detection:

# Good
def forward(self, task, visited=None):
    if visited is None:
        visited = set()
    if self.name in visited:
        return
    visited.add(self.name)

❌ No error handling: System crashes

# Bad
result = agent1(task)
result2 = agent2(result.output)  # May fail!

✅ Handle errors:

# Good
try:
    result = agent1(task)
    result2 = agent2(result.output)
except Exception as e:
    return fallback_response()

---

Related Skills

• dspy-agents.md - Single agent patterns
• dspy-optimizers.md - GEPA and other optimizers
• dspy-production.md - Deploying multi-agent systems
• dspy-debugging.md - Debugging agent interactions
• dspy-testing.md - Testing multi-agent systems
• dspy-rag.md - Multi-agent RAG patterns

---

Last Updated: 2025-10-30 Format Version: 1.0 (Atomic)