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. Optional at construction time, but run() raises InvalidAlgorithmError if it is still None when called.

str

Unique identifier for the algorithm instance. Auto-generated as a UUID if omitted.

float

default:"300.0"

Maximum execution time in seconds

OutputType

default:"dict"

Format for output (dict, list, str)

bool

default:"False"

Enable detailed logging

enable_communication_logging, parallel_execution and max_workers were removed. Agent messages are now always recorded into conversation, and the two parallel options were stored but never used. Passing them is still accepted and ignored, so existing call sites do not break.

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)

add_agent() / remove_agent()

Dynamic agent management.

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

remove_agent(agent_name: str) scans self.agents for a matching agent_name and deletes that entry. If no agent with that name exists, it raises AgentNotFoundError instead of silently no-op’ing.

social_alg.remove_agent("OldAgent")

get_communication_history()

Every agent message produced while the algorithm runs, oldest first. Recording is always on — there is no flag to enable.

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

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

# Each entry is a conversation message dict:
#   {"role": "<agent_name>", "content": "<the agent's answer>", ...}

The same messages are reachable through the standard Conversation API, which is usually the nicer way to read them:

print(social_alg.conversation.get_str())

for message in social_alg.conversation.conversation_history:
    print(message["role"], message["content"])

This returns message dicts, not CommunicationStep objects. That dataclass was removed when Conversation took over the transcript.

clear_communication_history()

Empties the recorded messages. The conversation otherwise accumulates across runs, so call this if you are reusing one instance for unrelated tasks.

SocialAlgorithmResult

Detailed execution results:

class SocialAlgorithmResult:
    algorithm_id: str
    execution_time: float
    total_steps: int
    successful_steps: int
    failed_steps: int
    communication_history: List[Dict[str, Any]]
    final_outputs: Any
    metadata: 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

Every agent call made while the algorithm runs is recorded automatically — the algorithm itself reports nothing. The task lands as User, each agent’s output under its own agent_name, and the final result under the algorithm’s name:

social_alg = SocialAlgorithms(
    name="Demo",
    agents=agents,
    social_algorithm=algorithm,
    verbose=True,
)

result = social_alg.run("Task")

for message in result.communication_history:
    print(f"{message['role']}: {str(message['content'])[:100]}")

User: Task
Researcher: Researcher handled: Research: Task
Analyst: Analyst handled: Analyze: ...
Demo: Analyst handled: Analyze: ...

talk_to calls record the receiver in the message metadata. The conversation accumulates across runs, so call clear_communication_history() between unrelated tasks.

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 (no dedicated get_agent_names() method exists;
# read agent_names from get_algorithm_info() or list agents directly)
names = [agent.agent_name for agent in social_alg.agents]

# 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()

add_agent() / remove_agent()

get_communication_history()

clear_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()

​add_agent() / remove_agent()

​get_communication_history()

​clear_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()

add_agent() / remove_agent()

get_communication_history()

clear_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