Orchestrating Multi-Agent Prompt Systems

So far, you’ve built pipelines where a single prompt handles each task. But complex problems often require multiple agents with different roles collaborating together. This lesson teaches you how to design and implement multi-agent systems where agents plan, execute, review, and improve each other’s work.

What Are Multi-Agent Systems?

A multi-agent system coordinates multiple specialized LLM instances, each with a specific role:

Research Agent          Writer Agent          Editor Agent
(Gathers Info)          (Creates Content)     (Improves Quality)
    ↓                       ↓                       ↓
    └───────────────────────┴───────────────────────┘
                        Orchestrator
                    (Manages Flow & State)

Why use multiple agents instead of one big prompt?

1. Separation of concerns: Each agent focuses on one task
2. Specialization: Agents can be optimized for their role
3. Iterative improvement: Agents refine each other’s work
4. Testability: Test agents independently
5. Transparency: See reasoning at each step
6. Scalability: Add agents without touching others

Defining Agent Roles

Before coding, define what each agent does:

from dataclasses import dataclass
from enum import Enum
from typing import Optional

class AgentRole(Enum):
    """Agent responsibilities in the system."""
    PLANNER = "planner"        # Breaks down goals into tasks
    EXECUTOR = "executor"      # Performs actual work
    REVIEWER = "reviewer"      # Checks quality
    CRITIC = "critic"          # Points out weaknesses
    SYNTHESIZER = "synthesizer" # Combines outputs

@dataclass
class AgentDefinition:
    """Configuration for an agent."""
    name: str
    role: AgentRole
    system_prompt: str
    model: str = "gpt-4"
    temperature: float = 0.7
    max_tokens: int = 1000

# Example: Define agents for content creation
CONTENT_CREATION_AGENTS = {
    "researcher": AgentDefinition(
        name="Research Agent",
        role=AgentRole.PLANNER,
        system_prompt="""You are a research agent. Your job is to:
1. Gather relevant information on the given topic
2. Organize facts logically
3. Identify knowledge gaps
4. Suggest areas for deeper exploration

Be thorough but concise.""",
        temperature=0.7
    ),

    "writer": AgentDefinition(
        name="Writer Agent",
        role=AgentRole.EXECUTOR,
        system_prompt="""You are a professional writer. Your job is to:
1. Create engaging, well-structured content
2. Use the research provided by other agents
3. Follow established tone and style
4. Include relevant examples and metaphors

Write in second person when appropriate.""",
        temperature=0.8
    ),

    "editor": AgentDefinition(
        name="Editor Agent",
        role=AgentRole.REVIEWER,
        system_prompt="""You are a meticulous editor. Your job is to:
1. Check for clarity and consistency
2. Improve flow and readability
3. Fix grammar and style issues
4. Ensure technical accuracy
5. Provide specific, actionable feedback

Be constructive but honest about issues.""",
        temperature=0.5
    ),

    "critic": AgentDefinition(
        name="Critic Agent",
        role=AgentRole.CRITIC,
        system_prompt="""You are a critical reviewer. Your job is to:
1. Point out weak arguments or unsupported claims
2. Identify missing perspectives
3. Suggest counterarguments
4. Highlight assumptions that may not hold
5. Recommend improvements

Be specific and evidence-based.""",
        temperature=0.6
    )
}

Communication Patterns Between Agents

Agents need structured ways to communicate:

from dataclasses import dataclass
from typing import List, Dict, Any
from datetime import datetime

@dataclass
class Message:
    """Message passed between agents."""
    sender: str           # Agent name
    recipient: str        # Target agent or "orchestrator"
    content: str          # Message body
    message_type: str     # "request", "response", "feedback", "summary"
    timestamp: datetime
    metadata: Dict[str, Any] = None

    def to_dict(self) -> dict:
        return {
            "sender": self.sender,
            "recipient": self.recipient,
            "content": self.content,
            "type": self.message_type,
            "timestamp": self.timestamp.isoformat()
        }

class AgentCommunicationChannel:
    """Manages communication between agents."""

    def __init__(self):
        self.message_log = []
        self.agent_states = {}

    def send_message(self, message: Message):
        """Send a message and log it."""
        self.message_log.append(message)
        print(f"[{message.sender} → {message.recipient}] {message.message_type}")

    def get_conversation_history(self, agent_name: str) -> List[Message]:
        """Get all messages involving an agent."""
        return [
            m for m in self.message_log
            if m.sender == agent_name or m.recipient == agent_name
        ]

    def get_agent_context(self, agent_name: str) -> str:
        """Get context from previous messages for an agent."""
        history = self.get_conversation_history(agent_name)

        if not history:
            return "No previous context."

        context = "Previous context:\n"
        for msg in history[-5:]:  # Last 5 messages
            context += f"\n{msg.sender}: {msg.content[:200]}..."

        return context

Building a Simple Multi-Agent System

Let’s implement a basic multi-agent orchestrator:

import openai
from typing import Optional

class Agent:
    """An individual agent in the system."""

    def __init__(self, definition: AgentDefinition, channel: AgentCommunicationChannel):
        self.definition = definition
        self.channel = channel
        self.message_count = 0

    def think(self, prompt: str, context: Optional[str] = None) -> str:
        """
        Agent thinks about the prompt using LLM.
        """
        # Build full message with context
        full_message = prompt
        if context:
            full_message = f"{context}\n\nTask: {prompt}"

        try:
            response = openai.ChatCompletion.create(
                model=self.definition.model,
                messages=[
                    {
                        "role": "system",
                        "content": self.definition.system_prompt
                    },
                    {
                        "role": "user",
                        "content": full_message
                    }
                ],
                temperature=self.definition.temperature,
                max_tokens=self.definition.max_tokens
            )

            thought = response.choices[0].message.content
            self.message_count += 1
            return thought

        except Exception as e:
            return f"Error: {str(e)}"

    def respond_to(self, recipient: str, prompt: str, message_type: str = "response"):
        """Think and send response to another agent or orchestrator."""
        thought = self.think(prompt)

        message = Message(
            sender=self.definition.name,
            recipient=recipient,
            content=thought,
            message_type=message_type,
            timestamp=datetime.now()
        )

        self.channel.send_message(message)
        return thought

class MultiAgentOrchestrator:
    """
    Coordinates multiple agents working together.
    Manages the flow of work between agents.
    """

    def __init__(self, agents_config: Dict[str, AgentDefinition]):
        self.channel = AgentCommunicationChannel()
        self.agents = {
            name: Agent(config, self.channel)
            for name, config in agents_config.items()
        }
        self.execution_log = []

    def execute_phase(self,
                     agent_name: str,
                     task: str,
                     phase_name: str) -> str:
        """
        Execute a single phase where one agent works.
        """
        agent = self.agents[agent_name]

        print(f"\n=== {phase_name} ({agent.definition.name}) ===")

        # Get context from previous phases
        context = self.channel.get_agent_context(agent_name)

        # Agent works on the task
        result = agent.respond_to("orchestrator", task, message_type="phase_output")

        self.execution_log.append({
            "phase": phase_name,
            "agent": agent_name,
            "task": task,
            "result": result
        })

        return result

    def get_execution_summary(self) -> dict:
        """Summarize the execution."""
        return {
            "total_phases": len(self.execution_log),
            "agents_used": list(self.agents.keys()),
            "total_messages": len(self.channel.message_log),
            "message_counts": {
                name: agent.message_count
                for name, agent in self.agents.items()
            }
        }

# Usage: Content creation workflow
orchestrator = MultiAgentOrchestrator(CONTENT_CREATION_AGENTS)

topic = "Prompt Engineering Best Practices"

# Phase 1: Research
research = orchestrator.execute_phase(
    "researcher",
    f"Research the topic: {topic}. Gather key facts and insights.",
    "Phase 1: Research"
)

# Phase 2: Writing
writing = orchestrator.execute_phase(
    "writer",
    f"Based on this research: {research}\n\nWrite an engaging article about {topic}",
    "Phase 2: Writing"
)

# Phase 3: Review
review = orchestrator.execute_phase(
    "editor",
    f"Review and improve this article: {writing}",
    "Phase 3: Review"
)

# Phase 4: Criticism
criticism = orchestrator.execute_phase(
    "critic",
    f"Critically review this article and suggest improvements: {review}",
    "Phase 4: Criticism"
)

print("\nExecution Summary:")
print(orchestrator.get_execution_summary())

Consensus and Voting Mechanisms

When multiple agents disagree, use voting or consensus:

from collections import Counter

class ConsensusResolver:
    """Resolve disagreements between agents using voting."""

    @staticmethod
    def simple_vote(agent_responses: Dict[str, str]) -> str:
        """
        Extract classifications and use majority vote.
        """
        # Parse classifications from responses
        votes = {}
        for agent_name, response in agent_responses.items():
            # Extract classification (assume it's the first line)
            classification = response.split('\n')[0].strip()
            votes[agent_name] = classification

        # Count votes
        vote_counts = Counter(votes.values())
        winner = vote_counts.most_common(1)[0][0]

        return {
            "consensus": winner,
            "votes": votes,
            "vote_distribution": dict(vote_counts)
        }

    @staticmethod
    def weighted_vote(agent_responses: Dict[str, str],
                     weights: Dict[str, float]) -> str:
        """
        Use weighted voting (some agents' opinions matter more).
        """
        weighted_score = {}

        for agent_name, response in agent_responses.items():
            classification = response.split('\n')[0].strip()
            weight = weights.get(agent_name, 1.0)

            if classification not in weighted_score:
                weighted_score[classification] = 0

            weighted_score[classification] += weight

        winner = max(weighted_score, key=weighted_score.get)

        return {
            "consensus": winner,
            "weighted_scores": weighted_score,
            "winning_score": weighted_score[winner]
        }

# Usage: Multiple agents classify something
classifier_agents = {
    "expert_classifier": CONTENT_CREATION_AGENTS["editor"],
    "critical_classifier": CONTENT_CREATION_AGENTS["critic"],
    "writer_classifier": CONTENT_CREATION_AGENTS["writer"]
}

orchestrator = MultiAgentOrchestrator(classifier_agents)

text_to_classify = "This product is amazing and I recommend it to everyone!"

responses = {}
for agent_name, agent_def in classifier_agents.items():
    agent = orchestrator.agents[agent_name]
    response = agent.think(f"Classify sentiment: {text_to_classify}")
    responses[agent_name] = response

# Resolve using weighted voting
weights = {
    "expert_classifier": 2.0,      # Expert opinion counts more
    "critical_classifier": 1.5,
    "writer_classifier": 1.0
}

consensus = ConsensusResolver.weighted_vote(responses, weights)
print(consensus)

Building Self-Reflecting Agents

Agents can review their own work and improve iteratively:

class ReflectiveAgent(Agent):
    """
    An agent that can reflect on its own output
    and try to improve it.
    """

    def reflect_and_improve(self, original_task: str,
                          original_output: str,
                          reflection_prompt: str = None) -> str:
        """
        Agent reviews its own work and suggests improvements.
        """
        if reflection_prompt is None:
            reflection_prompt = """Review this output critically:
{output}

What are the weaknesses? How could it be improved?"""

        reflection = self.think(
            reflection_prompt.format(output=original_output)
        )

        # Now use the reflection to improve
        improvement_prompt = f"""Based on this critique:
{reflection}

Original task: {original_task}
Original output: {original_output}

Please create an improved version."""

        improved_output = self.think(improvement_prompt)

        return {
            "original": original_output,
            "reflection": reflection,
            "improved": improved_output
        }

# Usage
reflective_agent = ReflectiveAgent(
    CONTENT_CREATION_AGENTS["writer"],
    AgentCommunicationChannel()
)

task = "Write a haiku about artificial intelligence"
initial = reflective_agent.think(task)
result = reflective_agent.reflect_and_improve(task, initial)

print("Original:", result["original"])
print("Reflection:", result["reflection"])
print("Improved:", result["improved"])

Key Takeaway: Multi-agent systems distribute tasks across specialized agents, enabling complex workflows that would be difficult for a single prompt. Communication channels, consensus mechanisms, and reflection create powerful collaborative systems.

Exercise: Build a 3-Agent Content Creation System

Create a system where three agents collaborate to produce high-quality content:

1. Researcher: Gathers information and outlines structure
2. Writer: Creates engaging content based on research
3. Editor: Reviews and provides improvements

Requirements:

• Define clear system prompts for each agent
• Implement structured communication between agents
• Use voting to resolve disagreements
• Track agent contributions
• Generate a final report showing agent interactions

Starter code:

class ContentCreationSystem:
    """Three-agent system for creating content."""

    def __init__(self):
        self.orchestrator = MultiAgentOrchestrator(CONTENT_CREATION_AGENTS)

    def create_content(self, topic: str, max_iterations: int = 3) -> dict:
        """
        Orchestrate agents to create and refine content.

        Returns:
            Dict with final content and agent contributions
        """
        # TODO: Phase 1 - Research
        # TODO: Phase 2 - Writing
        # TODO: Phase 3 - Review
        # TODO: Optional: Iterate for improvement
        # TODO: Return final content with metadata

        pass

system = ContentCreationSystem()
result = system.create_content("The Future of AI in Healthcare")

print("Final Content:", result["final_content"])
print("Agent Contributions:", result["contributions"])
print("Total Iterations:", result["iterations"])

Extension challenges:

• Add a quality evaluator agent that rates final output
• Implement iterative improvement loop (max 3 rounds)
• Add fact-checking agent to verify claims
• Create a visualization of agent interactions
• Implement cost tracking for multi-agent execution

By completing this exercise, you’ll understand how to orchestrate complex multi-agent workflows that tackle problems too difficult for single-prompt solutions.