SocialAlgorithms

pip install -U swarms

from swarms import SocialAlgorithms

def run(self, task: str, algorithm_args: Optional[Dict[str, Any]] = None, **kwargs) -> SocialAlgorithmResult

async def run_async(self, task: str, algorithm_args: Optional[Dict[str, Any]] = None, **kwargs) -> SocialAlgorithmResult

def add_agent(self, agent: Agent) -> None

def remove_agent(self, agent_name: str) -> None

def get_agent_names(self) -> List[str]

def get_communication_history(self) -> List[CommunicationStep]

def clear_communication_history(self) -> None

def get_algorithm_info(self) -> Dict[str, Any]

from swarms import Agent, SocialAlgorithms

# Define a custom social algorithm
def custom_communication_algorithm(agents, task, **kwargs):
    # Agent 1 researches the topic
    research_result = agents[0].run(f"Research: {task}")

    # Agent 2 analyzes the research
    analysis = agents[1].run(f"Analyze this research: {research_result}")

    # Agent 3 synthesizes the 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", model_name="gpt-4o-mini")
analyst = Agent(agent_name="Analyst", model_name="gpt-4o-mini")
synthesizer = Agent(agent_name="Synthesizer", model_name="gpt-4o-mini")

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

# Run the algorithm
result = social_alg.run("The impact of AI on healthcare")

from swarms import Agent, SocialAlgorithms

def research_development_algorithm(agents, task, **kwargs):
    """
    A comprehensive R&D team algorithm with multiple phases.
    """
    project_manager = next(a for a in agents if "ProjectManager" in a.agent_name)
    researcher = next(a for a in agents if "Researcher" in a.agent_name)
    analyst = next(a for a in agents if "Analyst" in a.agent_name)
    developer = next(a for a in agents if "Developer" in a.agent_name)
    tester = next(a for a in agents if "Tester" in a.agent_name)
    reviewer = next(a for a in agents if "Reviewer" in a.agent_name)

    # Phase 1: Project Planning
    project_plan = project_manager.run(f"""
    Create a comprehensive project plan for: {task}
    Include objectives, deliverables, timeline, and risk assessment.
    """)

    # Phase 2: Research
    research_findings = researcher.run(f"""
    Conduct comprehensive research on: {task}
    Cover current state of the art, existing solutions, and emerging trends.
    """)

    # Phase 3: Analysis and Design
    analysis_results = analyst.run(f"""
    Analyze the research findings and design the solution:
    Research: {research_findings}
    Include requirements analysis, architecture design, and implementation strategy.
    """)

    # Phase 4: Development
    prototype = developer.run(f"""
    Create a prototype based on the analysis:
    Analysis: {analysis_results}
    Requirements: {project_plan}
    """)

    # Phase 5: Testing
    test_results = tester.run(f"""
    Create test plans and execute testing:
    Prototype: {prototype}
    Requirements: {project_plan}
    """)

    # Phase 6: Final Review
    final_review = reviewer.run(f"""
    Conduct final review of the entire project:
    Plan: {project_plan}, Research: {research_findings},
    Analysis: {analysis_results}, Prototype: {prototype},
    Testing: {test_results}
    """)

    # Phase 7: Project Closure
    deliverables = project_manager.run(f"""
    Create final project deliverables including executive summary,
    documentation, and lessons learned.
    """)

    return {
        "task": task,
        "project_plan": project_plan,
        "research": research_findings,
        "analysis": analysis_results,
        "prototype": prototype,
        "test_results": test_results,
        "review": final_review,
        "deliverables": deliverables,
    }

# Create specialized agents
agents = [
    Agent(agent_name="ProjectManager", system_prompt="You are an experienced project manager.", model_name="gpt-4o-mini", max_loops=1),
    Agent(agent_name="Researcher", system_prompt="You are a research specialist.", model_name="gpt-4o-mini", max_loops=1),
    Agent(agent_name="Analyst", system_prompt="You are a systems analyst.", model_name="gpt-4o-mini", max_loops=1),
    Agent(agent_name="Developer", system_prompt="You are a senior developer.", model_name="gpt-4o-mini", max_loops=1),
    Agent(agent_name="Tester", system_prompt="You are a QA specialist.", model_name="gpt-4o-mini", max_loops=1),
    Agent(agent_name="Reviewer", system_prompt="You are a technical reviewer.", model_name="gpt-4o-mini", max_loops=1),
]

rd_algorithm = SocialAlgorithms(
    name="Research-Development-Team",
    description="Complete R&D workflow with specialized team members",
    agents=agents,
    social_algorithm=research_development_algorithm,
    verbose=True,
    max_execution_time=600
)

result = rd_algorithm.run("Develop a sustainable energy management system for smart cities")

from swarms import Agent, SocialAlgorithms

def competitive_evaluation_algorithm(agents, task, **kwargs):
    """Agents compete and are evaluated by a judge."""
    if len(agents) < 3:
        raise ValueError("Requires at least 3 agents (2 competitors + 1 judge)")

    competitors = agents[:-1]
    judge = agents[-1]

    # Each competitor works on the task
    competitor_results = {}
    for i, competitor in enumerate(competitors):
        result = competitor.run(f"Solve this task as best as you can: {task}")
        competitor_results[f"competitor_{i+1}_{competitor.agent_name}"] = result

    # Judge evaluates all solutions
    evaluation_prompt = "Evaluate these solutions and rank them:\n\n"
    for name, result in competitor_results.items():
        evaluation_prompt += f"{name}:\n{result}\n\n"
    evaluation_prompt += "Provide rankings, scores, and detailed feedback."

    evaluation = judge.run(evaluation_prompt)

    return {
        "competitor_solutions": competitor_results,
        "judge_evaluation": evaluation,
        "task": task,
    }

social_alg = SocialAlgorithms(
    name="Competitive-Evaluation",
    description="Competitive evaluation where agents compete and are judged",
    agents=[competitor1, competitor2, judge],
    social_algorithm=competitive_evaluation_algorithm,
    verbose=True,
)

result = social_alg.run("Design the most efficient algorithm for sorting large datasets")

from swarms import Agent, SocialAlgorithms

def negotiation_algorithm(agents, task, **kwargs):
    """Agents engage in back-and-forth negotiation with mediation."""
    negotiating_parties = agents[:-2]
    mediator_agent = agents[-2]
    legal_agent = agents[-1]

    max_rounds = kwargs.get("max_rounds", 8)

    # Phase 1: Initial Position Statements
    current_positions = {}
    for party in negotiating_parties:
        initial_position = party.run(
            f"As {party.agent_name}, state your initial position for: {task}"
        )
        current_positions[party.agent_name] = initial_position

    # Phase 2: Negotiation Rounds
    negotiation_history = []
    for round_num in range(1, max_rounds + 1):
        mediation_guidance = mediator_agent.run(
            f"Analyze positions for round {round_num}"
        )

        round_responses = {}
        for party in negotiating_parties:
            response = party.run(
                f"Respond to other positions in round {round_num}"
            )
            round_responses[party.agent_name] = response
            current_positions[party.agent_name] = response

        legal_review = legal_agent.run(
            f"Review proposals for round {round_num}"
        )

        negotiation_history.append({
            "round": round_num,
            "mediation_guidance": mediation_guidance,
            "responses": round_responses,
            "legal_review": legal_review,
        })

    return {
        "task": task,
        "negotiation_history": negotiation_history,
        "current_positions": current_positions,
    }