Sequential Pattern

Execute agents one after another in a linear pipeline, where each agent processes the output from the previous agent.

Overview

Sequential workflow is the simplest pattern, ideal for tasks that must be performed in order. Each agent receives the complete history of previous agent outputs and builds upon them.

When to Use

• Linear pipelines: Tasks that must be done in specific order
• Step-by-step transformations: Each step refines the previous output
• Simple automation: Straightforward, predictable workflows
• Clear dependencies: Each task depends on the previous one

Basic Usage

from azcore import SequentialWorkflow

# Create agents
writer = Agent(
    agent_name="Writer",
    system_prompt="Write a first draft",
    llm=llm,
)

editor = Agent(
    agent_name="Editor",
    system_prompt="Improve the draft's structure and clarity",
    llm=llm,
)

proofreader = Agent(
    agent_name="Proofreader",
    system_prompt="Fix grammar and spelling",
    llm=llm,
)

# Create sequential workflow
workflow = SequentialWorkflow(
    agents=[writer, editor, proofreader],
    max_loops=1,
)

# Run the workflow
result = workflow.run("Write an article about AI")
print(result)

Configuration Options

max_loops

Controls how many times the entire sequence runs:

workflow = SequentialWorkflow(
    agents=[agent1, agent2, agent3],
    max_loops=2,  # Run the full sequence twice
)

verbose

Enable detailed logging:

workflow = SequentialWorkflow(
    agents=[agent1, agent2],
    verbose=True,  # Log each agent's execution
)

return_full_history

Return complete conversation history:

workflow = SequentialWorkflow(
    agents=[agent1, agent2],
    return_full_history=True,  # Include all intermediate outputs
)

Advanced Examples

Content Pipeline

from azcore import Agent, SequentialWorkflow

# Research phase
researcher = Agent(
    agent_name="Researcher",
    system_prompt="Research the topic and gather key facts",
    llm=llm,
    tools=[search_tool, web_scraper],
)

# Writing phase
writer = Agent(
    agent_name="Writer",
    system_prompt="Create engaging content using the research",
    llm=llm,
)

# Editing phase
editor = Agent(
    agent_name="Editor",
    system_prompt="Improve structure, clarity, and flow",
    llm=llm,
)

# SEO optimization
seo_optimizer = Agent(
    agent_name="SEO",
    system_prompt="Optimize for search engines while maintaining quality",
    llm=llm,
    tools=[keyword_analyzer],
)

# Create pipeline
content_pipeline = SequentialWorkflow(
    agents=[researcher, writer, editor, seo_optimizer],
    max_loops=1,
    verbose=True,
)

# Generate article
article = content_pipeline.run(
    "Write an SEO-optimized article about renewable energy"
)

Code Development Workflow

# Design phase
architect = Agent(
    agent_name="Architect",
    system_prompt="Design system architecture and API",
    llm=llm,
)

# Implementation phase
developer = Agent(
    agent_name="Developer",
    system_prompt="Implement the code following the design",
    llm=llm,
    tools=[code_execution_tool],
)

# Testing phase
tester = Agent(
    agent_name="Tester",
    system_prompt="Write and run comprehensive tests",
    llm=llm,
    tools=[pytest_tool],
)

# Documentation phase
documenter = Agent(
    agent_name="Documenter",
    system_prompt="Create clear documentation and examples",
    llm=llm,
)

# Create development workflow
dev_workflow = SequentialWorkflow(
    agents=[architect, developer, tester, documenter],
    max_loops=1,
)

# Develop feature
result = dev_workflow.run("Create a user authentication system")

Multi-Stage Data Processing

# Data collection
collector = Agent(
    agent_name="Collector",
    system_prompt="Collect data from various sources",
    llm=llm,
    tools=[api_tools, database_tools],
)

# Data cleaning
cleaner = Agent(
    agent_name="Cleaner",
    system_prompt="Clean and normalize the data",
    llm=llm,
    tools=[pandas_tool],
)

# Data analysis
analyzer = Agent(
    agent_name="Analyzer",
    system_prompt="Perform statistical analysis",
    llm=llm,
    tools=[analysis_tools],
)

# Visualization
visualizer = Agent(
    agent_name="Visualizer",
    system_prompt="Create clear visualizations",
    llm=llm,
    tools=[plotting_tools],
)

# Report generation
reporter = Agent(
    agent_name="Reporter",
    system_prompt="Generate comprehensive report",
    llm=llm,
)

# Create data pipeline
data_pipeline = SequentialWorkflow(
    agents=[collector, cleaner, analyzer, visualizer, reporter],
    max_loops=1,
    verbose=True,
)

# Run analysis
report = data_pipeline.run("Analyze sales data for Q4 2024")

State Management

Each agent receives the complete conversation history:

# Agent 1 output becomes input for Agent 2
# Agent 2 sees: [task, agent1_output]
# Agent 3 sees: [task, agent1_output, agent2_output]

Access state explicitly:

class CustomSequentialWorkflow(SequentialWorkflow):
    def process_agent_output(self, agent_index, output, state):
        # Custom state processing
        state["processed_outputs"].append(output)
        return state

Error Handling

Handle failures gracefully:

from azcore import SequentialWorkflow

try:
    result = workflow.run("Task description")
except Exception as e:
    print(f"Workflow failed: {e}")
    # Implement fallback logic

Best Practices

1. Order Matters

Arrange agents logically:

# Good: Logical progression
[researcher, writer, editor, proofreader]

# Bad: Illogical order
[proofreader, writer, researcher, editor]

2. Keep Agents Focused

Each agent should have a clear, specific role:

# Good: Focused agents
writer = Agent(system_prompt="Write engaging content")
editor = Agent(system_prompt="Improve structure and clarity")

# Bad: Overlapping roles
writer = Agent(system_prompt="Write and edit content")

3. Limit Sequence Length

Shorter sequences are easier to debug:

# Good: 3-5 agents
workflow = SequentialWorkflow(agents=[agent1, agent2, agent3])

# Consider splitting: 10+ agents
# Maybe use hierarchical or forest patterns instead

4. Use Appropriate max_loops

# Single pass for most tasks
workflow = SequentialWorkflow(agents=[...], max_loops=1)

# Multiple passes for refinement
workflow = SequentialWorkflow(agents=[...], max_loops=3)

Performance Considerations

Latency

Sequential workflows have cumulative latency:

Total time = Agent1 + Agent2 + Agent3 + ...

For faster execution, consider Concurrent Workflow if tasks are independent.

Memory Usage

Each agent sees full history, which grows linearly:

# For long sequences, consider truncating history
# or using summarization agents periodically

Comparison with Other Patterns

Debugging

Enable verbose mode for detailed logs:

workflow = SequentialWorkflow(
    agents=[agent1, agent2, agent3],
    verbose=True,  # See each step
    return_full_history=True,  # Inspect all outputs
)

result = workflow.run("Task")

# Inspect history
for i, output in enumerate(result["history"]):
    print(f"Agent {i}: {output}")