SequentialWorkflow

pip install -U swarms

class SequentialWorkflow:
    def __init__(
        self,
        id: str = "sequential_workflow",
        name: str = "SequentialWorkflow",
        description: str = "Sequential Workflow, where agents are executed in a sequence.",
        agents: List[Union[Agent, Callable]] = None,
        max_loops: int = 1,
        output_type: OutputType = "dict",
        shared_memory_system: callable = None,
        multi_agent_collab_prompt: bool = False,
        team_awareness: bool = False,
        autosave: bool = True,
        verbose: bool = False,
        drift_detection: bool = False,
        drift_threshold: float = 0.75,
        drift_model: str = "claude-sonnet-4-5",
        *args,
        **kwargs,
    )

for token in workflow.run_stream("Analyse NVDA"):
    print(token, end="", flush=True)

async for token in workflow.arun_stream("Analyse NVDA"):
    print(token, end="", flush=True)

async for evt in workflow.arun_stream(task, with_events=True):
    if evt["type"] == "agent_start":
        print(f"--- {evt['agent']} starting ---")
    elif evt["type"] == "token":
        print(evt["token"], end="", flush=True)
    elif evt["type"] == "agent_end":
        print(f"\n--- {evt['agent']} finished ({len(evt['output'])} chars) ---")

from swarms import Agent, SequentialWorkflow

# Create specialized agents
researcher = Agent(
    agent_name="Researcher",
    model_name="claude-sonnet-4-6",
    temperature=0.5,
    max_loops=1,
    system_prompt="You are a research specialist. Gather and analyze information."
)

writer = Agent(
    agent_name="Writer",
    model_name="claude-sonnet-4-6",
    temperature=0.5,
    max_loops=1,
    system_prompt="You are a content writer. Create engaging content based on research."
)

editor = Agent(
    agent_name="Editor",
    model_name="claude-sonnet-4-6",
    temperature=0.5,
    max_loops=1,
    system_prompt="You are an editor. Review and polish the content."
)

# Create sequential workflow
workflow = SequentialWorkflow(
    name="Content Creation Pipeline",
    description="Research -> Write -> Edit pipeline",
    agents=[researcher, writer, editor],
    max_loops=1,
    verbose=True
)

# Run the workflow
result = workflow.run("Create a blog post about artificial intelligence")
print(result)

# Process multiple tasks through the same workflow
tasks = [
    "Analyze the impact of AI on healthcare",
    "Discuss the future of renewable energy",
    "Examine cybersecurity trends"
]

results = workflow.run_batched(tasks)

for task, result in zip(tasks, results):
    print(f"Task: {task}")
    print(f"Result: {result}")
    print("-" * 50)

import asyncio

# Run multiple tasks concurrently
tasks = [
    "Research quantum computing",
    "Research blockchain technology",
    "Research machine learning"
]

results = workflow.run_concurrent(tasks)
print(f"Processed {len(results)} tasks concurrently")

import asyncio

async def process_task():
    result = await workflow.run_async("Analyze climate change solutions")
    return result

# Run async
result = asyncio.run(process_task())
print(result)

# Create workflow with team awareness
workflow = SequentialWorkflow(
    name="Collaborative Pipeline",
    agents=[researcher, writer, editor],
    team_awareness=True,  # Agents know about each other
    multi_agent_collab_prompt=True,  # Add collaboration prompt
    verbose=True
)

result = workflow.run("Create a comprehensive market analysis")

# Different output formats
workflow_dict = SequentialWorkflow(
    agents=[researcher, writer],
    output_type="dict"  # Returns dict format
)

workflow_list = SequentialWorkflow(
    agents=[researcher, writer],
    output_type="list"  # Returns list format
)

workflow_str = SequentialWorkflow(
    agents=[researcher, writer],
    output_type="str"  # Returns string format
)

import os

# Set workspace directory for autosave
os.environ["WORKSPACE_DIR"] = "./my_workspace"

workflow = SequentialWorkflow(
    name="SavedWorkflow",
    agents=[researcher, writer, editor],
    autosave=True,  # Automatically save conversation history
    verbose=True
)

result = workflow.run("Generate a report")
# Conversation history saved to ./my_workspace/swarms/SequentialWorkflow/

try:
    workflow = SequentialWorkflow(
        agents=[researcher, writer],
        max_loops=1
    )
    result = workflow.run("Process this task")
except ValueError as e:
    print(f"Configuration error: {e}")
except Exception as e:
    print(f"Execution error: {e}")