Social Algorithms

The SocialAlgorithms framework provides complete flexibility for defining custom communication patterns between agents. Upload any arbitrary social algorithm as a callable that defines exactly how agents interact and communicate.

When to Use

Custom communication patterns: Unique agent interaction requirements
Research implementations: Test novel multi-agent algorithms
Specialized workflows: Domain-specific communication protocols
Flexible orchestration: Full control over agent interactions
Algorithm experimentation: Compare different social patterns

Key Features

Accept any callable as social algorithm
Communication history tracking (optional)
Execution timeout management
Multiple output formats
Agent lifecycle management
Async execution support
Parallel execution options
Detailed logging and monitoring

Basic Example

from swarms import Agent, SocialAlgorithms

# Define custom social algorithm
def research_analysis_synthesis(agents, task, **kwargs):
    """
    Custom pattern: Research -> Analysis -> Synthesis
    """
    # Agent 0: Research the topic
    research_result = agents[0].run(f"Research: {task}")
    
    # Agent 1: Analyze the research
    analysis = agents[1].run(f"Analyze this research: {research_result}")
    
    # Agent 2: Synthesize findings
    synthesis = agents[2].run(f"Synthesize: {research_result} + {analysis}")
    
    return {
        "research": research_result,
        "analysis": analysis,
        "synthesis": synthesis
    }

# Create agents
researcher = Agent(
    agent_name="Researcher",
    system_prompt="Expert in research and information gathering.",
    model_name="gpt-5.4",
)

analyst = Agent(
    agent_name="Analyst",
    system_prompt="Specialist in analyzing and interpreting data.",
    model_name="gpt-5.4",
)

synthesizer = Agent(
    agent_name="Synthesizer",
    system_prompt="Expert in synthesizing insights.",
    model_name="gpt-5.4",
)

# Create social algorithm
social_alg = SocialAlgorithms(
    name="Research-Analysis-Synthesis",
    agents=[researcher, analyst, synthesizer],
    social_algorithm=research_analysis_synthesis,
    verbose=True,
)

# Execute
result = social_alg.run("Impact of AI on healthcare")
print(result.final_outputs)

Custom Algorithm Patterns

Debate Algorithm

def debate_algorithm(agents, task, rounds=3, **kwargs):
    """
    Two agents debate with a judge evaluating.
    """
    pro_agent, con_agent, judge = agents[0], agents[1], agents[2]
    
    debate_history = []
    
    for round in range(rounds):
        # Pro argument
        pro_arg = pro_agent.run(f"Round {round+1} - Argue for: {task}")
        debate_history.append({"round": round+1, "pro": pro_arg})
        
        # Con argument
        con_arg = con_agent.run(f"Round {round+1} - Argue against: {task}. Pro said: {pro_arg}")
        debate_history.append({"round": round+1, "con": con_arg})
    
    # Judge evaluation
    judgment = judge.run(f"Evaluate debate: {debate_history}")
    
    return {
        "debate_history": debate_history,
        "judgment": judgment
    }

social_alg = SocialAlgorithms(
    name="Debate-System",
    agents=[pro_agent, con_agent, judge],
    social_algorithm=debate_algorithm,
)

Hierarchical Review

def hierarchical_review_algorithm(agents, task, **kwargs):
    """
    Draft -> Peer Review -> Senior Review -> Approval
    """
    drafter, peer, senior, approver = agents
    
    # Initial draft
    draft = drafter.run(f"Create draft: {task}")
    
    # Peer review
    peer_review = peer.run(f"Review this draft: {draft}")
    
    # Revision based on peer review
    revision = drafter.run(f"Revise draft based on: {peer_review}")
    
    # Senior review
    senior_review = senior.run(f"Senior review of: {revision}")
    
    # Final approval
    approval = approver.run(f"Approve or reject: {senior_review}")
    
    return {
        "draft": draft,
        "peer_review": peer_review,
        "revision": revision,
        "senior_review": senior_review,
        "approval": approval
    }

Consensus Building

def consensus_algorithm(agents, task, threshold=0.7, **kwargs):
    """
    Agents vote and iterate until consensus reached.
    """
    max_iterations = 5
    proposals = {}
    
    for iteration in range(max_iterations):
        # Each agent makes a proposal
        for agent in agents:
            context = f"Previous proposals: {proposals}" if proposals else ""
            proposal = agent.run(f"{task}. {context}")
            proposals[agent.agent_name] = proposal
        
        # Check for consensus (simplified)
        # In real implementation, measure similarity
        if iteration >= 2:  # At least 3 rounds
            break
    
    # Final synthesis
    synthesizer = agents[0]
    final = synthesizer.run(f"Synthesize consensus from: {proposals}")
    
    return {
        "proposals": proposals,
        "consensus": final
    }

Key Parameters

str

default:"SocialAlgorithm"

Name for the algorithm instance

str

Description of the algorithm’s purpose

List[Agent]

required

List of agents that will participate

Function defining the communication pattern (agents, task, **kwargs) -> Any

float

default:"300.0"

Maximum execution time in seconds

OutputType

default:"dict"

Format for output (dict, list, str)

bool

default:"False"

Log all agent communications

bool

default:"False"

Enable parallel execution where possible

bool

default:"False"

Enable detailed logging

Methods

run()

Execute the social algorithm.

result = social_alg.run(
    task="Analyze market trends",
    algorithm_args={"rounds": 3, "threshold": 0.8},  # Custom args
)

# Result is SocialAlgorithmResult object
print(result.final_outputs)
print(result.execution_time)
print(result.successful_steps)

run_async()

Asynchronous execution.

import asyncio

result = social_alg.run_async(
    task="Task description",
    algorithm_args={"custom_param": "value"},
)

add_agent() / remove_agent()

Dynamic agent management.

# Add new agent
new_agent = Agent(agent_name="NewExpert", ...)
social_alg.add_agent(new_agent)

# Remove agent
social_alg.remove_agent("OldAgent")

get_communication_history()

Retrieve communication logs (if enabled).

social_alg = SocialAlgorithms(
    agents=agents,
    social_algorithm=algorithm,
    enable_communication_logging=True,
)

result = social_alg.run("Task")
history = social_alg.get_communication_history()

# Each step: CommunicationStep(
#   sender_agent, receiver_agent, message, timestamp
# )

SocialAlgorithmResult

Detailed execution results:

class SocialAlgorithmResult:
    algorithm_id: str
    execution_time: float
    total_steps: int
    successful_steps: int
    failed_steps: int
    communication_history: List[CommunicationStep]
    final_outputs: Dict[str, Any]
    metadata: Optional[Dict[str, Any]]

# Access results
result = social_alg.run("Task")
print(f"Execution took {result.execution_time:.2f}s")
print(f"Successful steps: {result.successful_steps}")
print(f"Output: {result.final_outputs}")

Use Cases

Multi-Stage Pipeline

def pipeline_algorithm(agents, task, **kwargs):
    collector, cleaner, analyzer, reporter = agents
    
    # Stage 1: Data collection
    data = collector.run(f"Collect data for: {task}")
    
    # Stage 2: Data cleaning
    clean_data = cleaner.run(f"Clean this data: {data}")
    
    # Stage 3: Analysis
    analysis = analyzer.run(f"Analyze: {clean_data}")
    
    # Stage 4: Report generation
    report = reporter.run(f"Create report from: {analysis}")
    
    return {"report": report, "analysis": analysis}

pipeline = SocialAlgorithms(
    name="Data-Pipeline",
    agents=[collector, cleaner, analyzer, reporter],
    social_algorithm=pipeline_algorithm,
)

Collaborative Writing

def collaborative_writing_algorithm(agents, task, **kwargs):
    outline_agent, section_agents, editor = agents[0], agents[1:-1], agents[-1]
    
    # Create outline
    outline = outline_agent.run(f"Create outline for: {task}")
    
    # Each agent writes a section
    sections = []
    for i, agent in enumerate(section_agents):
        section = agent.run(f"Write section {i+1} based on outline: {outline}")
        sections.append(section)
    
    # Editor combines and polishes
    final = editor.run(f"Combine and edit sections: {sections}")
    
    return {"outline": outline, "sections": sections, "final": final}

writing_team = SocialAlgorithms(
    agents=[outliner, writer1, writer2, writer3, editor],
    social_algorithm=collaborative_writing_algorithm,
)

Expert Panel

def expert_panel_algorithm(agents, task, **kwargs):
    """
    All experts analyze, then discuss findings.
    """
    # Phase 1: Individual analysis
    individual_analyses = []
    for expert in agents:
        analysis = expert.run(f"Analyze: {task}")
        individual_analyses.append({
            "expert": expert.agent_name,
            "analysis": analysis
        })
    
    # Phase 2: Discussion
    discussion = []
    for expert in agents:
        context = f"Other experts said: {individual_analyses}"
        response = expert.run(f"Discuss and respond: {context}")
        discussion.append({"expert": expert.agent_name, "response": response})
    
    return {
        "individual_analyses": individual_analyses,
        "discussion": discussion
    }

expert_panel = SocialAlgorithms(
    agents=[expert1, expert2, expert3, expert4],
    social_algorithm=expert_panel_algorithm,
)

Communication Logging

Track all agent interactions:

social_alg = SocialAlgorithms(
    agents=agents,
    social_algorithm=algorithm,
    enable_communication_logging=True,
    verbose=True,
)

result = social_alg.run("Task")

# Access communication history
for step in result.communication_history:
    print(f"{step.sender_agent} -> {step.receiver_agent}")
    print(f"Message: {step.message[:100]}...")
    print(f"Timestamp: {step.timestamp}")

Execution Timeout

social_alg = SocialAlgorithms(
    agents=agents,
    social_algorithm=long_running_algorithm,
    max_execution_time=300.0,  # 5 minutes
)

try:
    result = social_alg.run("Complex task")
except TimeoutError:
    print("Algorithm execution exceeded timeout")

Output Formatting

# Dictionary output (default)
social_alg_dict = SocialAlgorithms(
    agents=agents,
    social_algorithm=algorithm,
    output_type="dict",
)

# List output
social_alg_list = SocialAlgorithms(
    agents=agents,
    social_algorithm=algorithm,
    output_type="list",
)

# String output
social_alg_str = SocialAlgorithms(
    agents=agents,
    social_algorithm=algorithm,
    output_type="str",
)

Algorithm Requirements

Your social algorithm must:

Accept agents and task: def algorithm(agents, task, **kwargs)
Return results: Any structure (dict, list, str, object)
Handle errors: Exceptions will be caught and logged

def valid_algorithm(agents, task, **kwargs):
    # Your logic here
    result = agents[0].run(task)
    return {"result": result}

def invalid_algorithm(wrong_params):
    # Missing agents and task parameters
    pass

Best Practices

Algorithm Design: Keep algorithms focused on communication patterns, not complex logic

Clear Signatures: Always accept (agents, task, **kwargs)
Error Handling: Handle agent failures gracefully
Timeout Awareness: Set appropriate max_execution_time
Communication Logging: Enable for debugging and analysis
Documentation: Document your algorithm’s communication pattern

Social algorithms have full control over agent execution - ensure proper error handling and timeout limits

Error Handling

try:
    result = social_alg.run("Task")
except InvalidAlgorithmError:
    print("Social algorithm is not callable")
except TimeoutError:
    print("Execution exceeded max_execution_time")
except Exception as e:
    print(f"Algorithm execution failed: {e}")

Agent Management

# Get agent names
names = social_alg.get_agent_names()

# Get algorithm info
info = social_alg.get_algorithm_info()
print(info)
# {
#   "algorithm_id": "...",
#   "name": "...",
#   "agent_count": 3,
#   "has_algorithm": True,
#   "max_execution_time": 300.0,
#   ...
# }

Agent Rearrange - Predefined flow patterns
Graph Workflow - DAG-based workflows
Group Chat - Conversational patterns
Hierarchical Swarm - Director-worker pattern

Getting Started

Core Concepts

Agent Development

Reasoning Agents

Multi-Agent Architectures

Deployment & Operations

Integrations

CLI Reference

Changelog

Social Algorithms

When to Use

Key Features

Basic Example

Custom Algorithm Patterns

Debate Algorithm

Hierarchical Review

Consensus Building

Key Parameters

Methods

run()

run_async()

add_agent() / remove_agent()

get_communication_history()

SocialAlgorithmResult

Use Cases

Multi-Stage Pipeline

Collaborative Writing

Expert Panel

Communication Logging

Execution Timeout

Output Formatting

Algorithm Requirements

Best Practices

Error Handling

Agent Management

​When to Use

​Key Features

​Basic Example

​Custom Algorithm Patterns

​Debate Algorithm

​Hierarchical Review

​Consensus Building

​Key Parameters

​Methods

​run()

​run_async()

​add_agent() / remove_agent()

​get_communication_history()

​SocialAlgorithmResult

​Use Cases

​Multi-Stage Pipeline

​Collaborative Writing

​Expert Panel

​Communication Logging

​Execution Timeout

​Output Formatting

​Algorithm Requirements

​Best Practices

​Error Handling

​Agent Management

​Related Architectures

When to Use

Key Features

Basic Example

Custom Algorithm Patterns

Debate Algorithm

Hierarchical Review

Consensus Building

Key Parameters

Methods

run()

run_async()

add_agent() / remove_agent()

get_communication_history()

SocialAlgorithmResult

Use Cases

Multi-Stage Pipeline

Collaborative Writing

Expert Panel

Communication Logging

Execution Timeout

Output Formatting

Algorithm Requirements

Best Practices

Error Handling

Agent Management

Related Architectures