Parallel Agent Execution with AG2: Best Practices Guide | AG2 Blog

Parallel Agent Execution with AG2: Best Practices Guide

Justin Trugman Cofounder & Head of Technology, BetterFutureLabs

·April 15, 2025·5 min read

Multi-threading with open-source LLMs was a game changer for us. We've been able to run independent tasks simultaneously, instead of waiting for agents sequentially, slashing turnaround times on document processing. Our team can now process 5x more cases without proportional headcount increases.

— Justin Trugman, Cofounder & Head of Technology, BetterFutureLabs

Why Parallel Execution Matters

Consider a real-world scenario where you're building a medical diagnosis support system. Your system needs multiple specialized multi-agent teams simultaneously analyzing different aspects of a patient case:

Each team must reach internal consensus through multi-agent discussions before contributing to the final diagnostic assessment.

Sequential execution: 40-58 minutes total (each team waits for the previous to complete)

Parallel execution: 10-15 minutes total (limited by the longest-running team)

This 70-75% reduction in processing time transforms diagnosis from a lengthy batch process into a near real-time clinical decision support tool.

Core Implementation Pattern

We've found success using Python's concurrent.futures.ThreadPoolExecutor as the foundation for parallel agent execution.

Why ThreadPoolExecutor works well for AG2:

import concurrent.futures

# Define independent diagnostic analysis tasks
diagnostic_tasks = [\
  "symptom_pattern_analysis",\
  "literature_review",\
  "drug_interaction_check",\
  "imaging_analysis",\
  "treatment_planning"\
]

# Execute diagnostic teams in parallel
with concurrent.futures.ThreadPoolExecutor() as executor:
  futures = [\
    executor.submit(run_diagnostic_team, team_id, analysis_type, patient_data)\
    for analysis_type in diagnostic_tasks\
  ]
  concurrent.futures.wait(futures)  # Wait for all teams to complete

Agent Factory Pattern

For parallel execution, each parallel execution creates completely fresh agent instances. This prevents state contamination and ensures true independence.

def run_diagnostic_team(team_id, analysis_type, patient_data):
  """
  Factory function that creates a complete independent agent team
  """
  # Create fresh user proxy for this diagnostic team
  user_proxy = UserProxyAgent(
    name=f"coordinator_{analysis_type}_{team_id}",
    is_termination_msg=lambda msg: "TERMINATE" in msg["content"],
    human_input_mode="NEVER",
    max_consecutive_auto_reply=1,
    code_execution_config=False
  )

# Create specialized medical analyst with unique configuration
  medical_specialist = GPTAssistantAgent(
    name=f"specialist_{analysis_type}_{team_id}",
    instructions=medical_instructions[analysis_type],
    overwrite_instructions=True,  # Ensure clean state
    overwrite_tools=True,
    llm_config={
      "config_list": config_list,
      "tools": diagnostic_tools[analysis_type],
      "assistant_id": specialist_assistant_ids[analysis_type]
    }
  )

# Create complete diagnostic team and execute
  team = [user_proxy, medical_specialist, clinical_reviewer, supervisor]
  groupchat = ag2.GroupChat(agents=team, messages=[], max_round=15)
  chat_manager = ag2.GroupChatManager(groupchat=groupchat, llm_config={"config_list": config_list})

user_proxy.initiate_chat(chat_manager, message=f"Perform {analysis_type} for patient: {patient_data}")

Key principles:

Common Pitfalls to Avoid

Race Conditions

The most dangerous trap is sharing mutable state between agents.

# BAD: Shared mutable state
shared_patient_analysis = {}
def bad_diagnostic_process(analysis_type):
  shared_patient_analysis[analysis_type] = perform_analysis(...)  # Race condition!

# GOOD: Independent storage
def good_diagnostic_process(analysis_type, patient_id):
  result = perform_analysis(...)
  save_to_file(f"diagnosis_{patient_id}_{analysis_type}.json", result)

Resource Exhaustion

Creating unlimited threads will overwhelm your system.

# BAD: Unlimited workers
with ThreadPoolExecutor() as executor:  # Could create too many threads

# GOOD: Controlled resource usage
max_workers = min(len(tasks), 5)
with ThreadPoolExecutor(max_workers=max_workers) as executor:

Ignoring Failures

In parallel execution, individual tasks will fail. Handle these gracefully.

# GOOD: Graceful error handling
for future in concurrent.futures.as_completed(futures):
  try:
    result = future.result(timeout=300)
    handle_success(result)
  except Exception as e:
    handle_error(e)

Production Implementation Tips

Error Handling

def execute_parallel_tasks(doc_id, tasks, document_data):
  results = {}
  errors = {}

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    future_to_task = {
      executor.submit(process_document, doc_id, task, document_data): task
      for task in tasks
    }

for future in concurrent.futures.as_completed(future_to_task):
      task = future_to_task[future]
      try:
        results[task] = future.result(timeout=300)
        logging.info(f"Task {task} completed successfully")
      except Exception as e:
        errors[task] = str(e)
        logging.error(f"Task {task} failed: {e}")

return results, errors

Resource Management

The default ThreadPoolExecutor worker count is min(32, (os.process_cpu_count() or 1) + 4).

import os
def get_optimal_workers(task_count):
  default_workers = min(32, (os.process_cpu_count() or 1) + 4)
  return min(task_count, default_workers)

Best Practices Summary

  1. Design for Complete Independence: Use separate data sources and storage for each parallel task
  2. Intelligent Resource Management: Calculate optimal workers, implement appropriate timeouts
  3. Comprehensive Error Handling: Graceful degradation, detailed logging
  4. Performance Monitoring: Track total execution time, individual task duration, success/failure rates

Expected outcomes when properly implemented:

Parallel execution fundamentally changes how users interact with multi-agent systems. Start with these patterns, measure your results, and iterate based on your specific use case requirements.