Skip to main content
Mapagam

Orchestrating Multi-Agent Workflows

1. Designing Orchestrator-Worker Patterns

Example: Basic Orchestrator-Worker Communication

// Orchestrator sends tasks to workers
const orchestrator = {
  workers: [],
  queue: [],
  assignTask(task) {
    const worker = this.workers.find(w => w.available);
    if (worker) {
      worker.execute(task);
    } else {
      this.queue.push(task);
    }
  }
};

// Worker receives and executes tasks
const worker = {
  available: true,
  execute(task) {
    this.available = false;
    // Process task
    setTimeout(() => {
      this.available = true;
    }, 1000);
  }
};
Pattern Description Use Case Benefits
Centralized Single orchestrator manages all workers Simple task distribution Easy to monitor and debug
Decentralized Workers coordinate among themselves Complex multi-step workflows Scalable and fault-tolerant
Hierarchical Multiple orchestrators manage sub-workers Large-scale distributed systems Modular and maintainable

Key Considerations

  • Task distribution strategies (round-robin, load balancing)
  • Worker availability and status tracking
  • Error handling and retry mechanisms
  • Communication protocols (REST, gRPC, message queues)

Orchestrator vs Worker Roles

  • Orchestrator: Task scheduling, coordination, monitoring
  • Worker: Task execution, result processing, status reporting

2. Implementing Sequential Agent Pipelines

Agent Role Input Output
DataCollector Fetches raw data from sources API endpoints, databases Raw data JSON
DataValidator Validates and sanitizes data Raw data JSON Validated data JSON
ModelRunner Runs ML models on validated data Validated data JSON Predictions JSON
ResultAggregator Combines predictions into final result Predictions JSON Final report JSON

Example: Sequential Pipeline Setup

const pipeline = [
  { name: 'DataCollector', run: collectData },
  { name: 'DataValidator', run: validateData },
  { name: 'ModelRunner', run: runModel },
  { name: 'ResultAggregator', run: aggregateResults }
];

async function executePipeline(data) {
  let result = data;
  for (const agent of pipeline) {
    console.log(`Running ${agent.name}`);
    result = await agent.run(result);
  }
  return result;
}

Example: DataCollector Agent

async function collectData(input) {
  const response = await fetch('https://api.example.com/data');
  const rawData = await response.json();
  return { type: 'raw', data: rawData };
}

Example: ModelRunner Agent

async function runModel(input) {
  const model = new MLModel();
  const predictions = await model.predict(input.data);
  return { type: 'predictions', data: predictions };
}
Note: Each agent in the pipeline should handle one specific task and pass its output to the next agent. This ensures modularity and maintainability.

3. Implementing Parallel Agent Fan-Out

Example: Fan-out Pattern with Concurrent Agents

const { Agent, Workflow } = require('ai-agent-framework');

const workflow = new Workflow();

// Define parallel agents
const agentA = new Agent({
  name: 'DataProcessorA',
  task: 'process data set A'
});

const agentB = new Agent({
  name: 'DataProcessorB',
  task: 'process data set B'
});

const agentC = new Agent({
  name: 'DataProcessorC',
  task: 'process data set C'
});

// Fan-out to all agents
workflow.fanOut([agentA, agentB, agentC], {
  onComplete: (results) => {
    console.log('All agents completed:', results);
  }
});
Property Description Values
fanOut Method to initiate parallel execution of agents function(agents, options)
agents Array of agent instances to execute in parallel Agent[]
options.onComplete Callback when all agents finish execution function(results)
Note: The fan-out pattern is ideal for scenarios where independent tasks can be executed simultaneously to reduce overall processing time.

4. Passing Context Between Agents

Context Type Description Implementation Notes
Shared Memory Agents share a common memory space to store and retrieve information const sharedMemory = new SharedMemory();\nagentA.writeToMemory("userIntent", "book a flight to Paris");\nconst intent = agentB.readFromMemory("userIntent"); Useful for maintaining state across multiple agent interactions
Message Passing Agents communicate by passing structured messages containing context const message = {\n type: "userIntent",\n content: "book a flight to Paris",\n metadata: { agentId: "agentA" }\n};\nagentB.receiveMessage(message); Ensures context is preserved and can be extended with metadata
State Propagation Context is propagated through a state object that agents can modify let state = {\n userQuery: "book a flight to Paris",\n destination: null\n};\nagentA.updateState(state, "destination", "Paris"); Allows for dynamic updates to shared context during workflow execution

Example: Agent Communication with Context

const contextManager = new ContextManager();\n\nconst agentA = new Agent({\n  name: "Planner",\n  onMessage: (message) => {\n    const context = contextManager.extract(message);\n    return { type: "intent", content: context.intent };\n  }\n});\n\nconst agentB = new Agent({\n  name: "Executor",\n  onMessage: (message) => {\n    const context = contextManager.extract(message);\n    return { type: "action", content: `Book flight to ${context.destination}` };\n  }\n});
Note: Context passing must be handled carefully to avoid information leakage or conflicts between agents.

5. Handling Agent Handoffs

Handoff Type Description Key Considerations
Direct Transfer Agent A directly passes control to Agent B with minimal context Ensure state consistency and prevent data loss during transfer
Contextual Handoff Agent A transfers control to Agent B with a detailed task summary and context Preserve task integrity and maintain shared understanding between agents
Decision-Based Handoff Agent A evaluates whether to hand off based on task complexity or resource constraints Define clear decision criteria and ensure agents can handle the handoff logic

Example: Contextual Handoff Implementation

class AgentHandoffManager {
  async handleHandoff(agentA, agentB, taskContext) {
    const handoffData = {
      taskId: taskContext.id,
      context: taskContext,
      timestamp: Date.now()
    };
    
    // Save current state before transfer
    await agentA.saveState(handoffData);
    
    // Transfer control to next agent with full context
    return await agentB.execute(handoffData);
  }
}

const handoffManager = new AgentHandoffManager();
const result = await handoffManager.handleHandoff(agentA, agentB, taskContext);
Note: Effective handoffs require a shared understanding of task goals and intermediate states to prevent rework or loss of progress.

Best Practices for Agent Handoffs

  • Define clear handoff triggers (e.g., task complexity, resource availability)
  • Maintain a consistent context format across agents
  • Implement rollback mechanisms for failed handoffs
  • Log all handoff events for debugging and analytics

6. Implementing Human-in-the-Loop Checkpoints

Example: Configuring a Human Review Step

const workflow = {
  steps: [
    { type: 'code_generation', agent: 'ai_agent_1' },
    {
      type: 'human_review',
      agent: 'human_reviewer',
      approval_required: true,
      timeout: 3600 // 1 hour
    },
    { type: 'code_execution', agent: 'executor' }
  ]
};
Property Description Default Value Notes
type Step type identifier n/a Must be human_review
agent Human reviewer identifier n/a Must be a valid human agent ID
approval_required Whether approval is mandatory false If true, workflow pauses for approval
timeout Timeout in seconds for human review 3600 Workflow will fail if not reviewed in time
Note: Human review steps can be used to ensure quality control in AI-assisted development pipelines.
Warning: If approval_required is set to true, the workflow will pause indefinitely until human approval is received.

7. Managing Shared State Across Agents

Approach Description Implementation Notes
Centralized State Store Use a shared database or in-memory store (e.g., Redis, DynamoDB) to manage state 
const redis = require('redis');
const client = redis.createClient();

await client.set('sharedState', JSON.stringify({ agent1: 'running', agent2: 'idle' }));
const state = await client.get('sharedState');
 Highly scalable, supports real-time updates
Message Queue Agents communicate via a message broker (e.g., RabbitMQ, Kafka) to synchronize state 
const amqp = require('amqplib');
const channel = await amqp.connect('amqp://localhost').then(conn => conn.createChannel());

channel.publish('stateUpdates', '', Buffer.from(JSON.stringify({ agentId: 'A1', status: 'completed' })));
 Decouples agents; supports asynchronous updates
Event Sourcing Record all state changes as events to replay or reconstruct state 
const events = [
  { type: 'agentStarted', agentId: 'A1', timestamp: Date.now() },
  { type: 'agentCompleted', agentId: 'A1', result: 'success' }
];

const currentState = events.reduce((state, event) => {
  if (event.type === 'agentStarted') state[event.agentId] = 'running';
  return state;
}, {});
 Provides audit trail and enables debugging

Example: Agent State Update via Redis

const redis = require('redis');
const client = redis.createClient();

// Simulate agent state update
async function updateAgentState(agentId, status) {
  const currentState = await client.get('sharedState');
  const state = JSON.parse(currentState || '{}');
  state[agentId] = status;
  await client.set('sharedState', JSON.stringify(state));
}

updateAgentState('A1', 'completed');
Note: Ensure atomicity when updating shared state to prevent race conditions.
Warning: Avoid tight coupling between agents and state stores; use abstraction layers for flexibility.

8. Resolving Conflicting Agent Outputs

Conflicting Output Type Description Resolution Strategy Example
Inconsistent API Responses Agents return different data structures from the same API call Normalize responses using a common schema or adapter pattern 
const normalized = {
  status: response.status,
  data: response.body.data || response.body.result
}
Disagreement on Task Completion Agents report different statuses for same task (e.g., "completed" vs "failed") Implement a consensus mechanism or voting system among agents 
const consensus = agents.map(a => a.status).reduce((acc, curr) => {
  return acc === curr ? acc : 'disagreement'
})
Conflicting Code Suggestions Multiple agents propose different implementations for same logic Use a code review agent or merge strategy to select best solution 
const bestCode = codeEvaluator.selectBest(
  agentA.suggestion,
  agentB.suggestion
)

Example: Conflict Resolution via Voting

const resolveConflict = (agentOutputs) => {
  const voteCounts = agentOutputs.reduce((acc, output) => {
    acc[output] = (acc[output] || 0) + 1;
    return acc;
  }, {});

  const maxVotes = Math.max(...Object.values(voteCounts));
  return Object.keys(voteCounts).find(key => voteCounts[key] === maxVotes);
};

const finalOutput = resolveConflict([
  "outputA",
  "outputB",
  "outputA"
]); // Returns "outputA"
Note: For critical workflows, implement a fallback mechanism if no clear consensus is reached.

Was this article helpful?

Related Cheat Sheets