Building Powerful Sequential Multi-Agent Systems: A Practical Guide

With the recent release of the Agents SDK by OpenAI and Google, I thought it would be fun to compare how the major frameworks implement sequential multi-agent systems for tackling multi-step workflows.

This blog post explores how to create effective sequential multi-agent systems using three popular frameworks: Google's Agent Development Kit (ADK), CrewAI, and OpenAI's Agents SDK.

Why Sequential Multi-Agent Systems Matter

Sequential multi-agent systems excel at breaking down complex tasks into manageable steps, with each step handled by a specialized agent. This approach offers several advantages:

• Modularity: Each agent focuses on what it does best
• Specialization: Agents can be optimized for specific tasks
• Maintainability: Easier to debug and improve individual components
• Reusability: Agents can be reused across different workflows
• Clarity: The workflow is explicit and easier to understand

These systems allow developers to create more modular, scalable, and understandable solutions that can tackle problems beyond just question answering.

Google ADK: Building a Sequential Code Pipeline

Google recently released its Agent Development Kit (ADK), a new open-source framework designed to simplify the full stack end-to-end development of agents and multi-agent systems. One of ADK's strengths is its flexible orchestration capabilities, including sequential workflows.

Here's how to implement a sequential code pipeline using Google's ADK official example:

from google.adk.agents.sequential_agent import SequentialAgent
from google.adk.agents.llm_agent import LlmAgent
from google.genai import types
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner

# Constants
APP_NAME = "code_pipeline_app"
USER_ID = "dev_user_01"
SESSION_ID = "pipeline_session_01"
GEMINI_MODEL = "gemini-2.0-flash"

# 1. Define Sub-Agents for Each Pipeline Stage
# Code Writer Agent
code_writer_agent = LlmAgent(
    name="CodeWriterAgent",
    model=GEMINI_MODEL,
    instruction="""You are a Code Writer AI.
    Based on the user's request, write the initial Python code.
    Output *only* the raw code block.
    """,
    description="Writes initial code based on a specification.",
    output_key="generated_code"
)

# Code Reviewer Agent
code_reviewer_agent = LlmAgent(
    name="CodeReviewerAgent",
    model=GEMINI_MODEL,
    instruction="""You are a Code Reviewer AI.
    Review the Python code provided in the session state under the key 'generated_code'.
    Provide constructive feedback on potential errors, style issues, or improvements.
    Focus on clarity and correctness.
    Output only the review comments.
    """,
    description="Reviews code and provides feedback.",
    output_key="review_comments"
)

# Code Refactorer Agent
code_refactorer_agent = LlmAgent(
    name="CodeRefactorerAgent",
    model=GEMINI_MODEL,
    instruction="""You are a Code Refactorer AI.
    Take the original Python code provided in the session state key 'generated_code'
    and the review comments found in the session state key 'review_comments'.
    Refactor the original code to address the feedback and improve its quality.
    Output *only* the final, refactored code block.
    """,
    description="Refactors code based on review comments.",
    output_key="refactored_code"
)

# 2. Create the SequentialAgent
code_pipeline_agent = SequentialAgent(
    name="CodePipelineAgent",
    sub_agents=[code_writer_agent, code_reviewer_agent, code_refactorer_agent]
    # The agents will run in the order provided: Writer -> Reviewer -> Refactorer
)

# Session and Runner
session_service = InMemorySessionService()
session = session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
runner = Runner(agent=code_pipeline_agent, app_name=APP_NAME, session_service=session_service)

# Agent Interaction
def call_agent(query):
    content = types.Content(role='user', parts=[types.Part(text=query)])
    events = runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=content)
    for event in events:
        if event.is_final_response():
            final_response = event.content.parts[0].text
            print("Agent Response: ", final_response)

# Example usage
call_agent("Create a function to calculate factorial recursively")

In this ADK example, we've created a sequential pipeline consisting of three specialized agents:

1. CodeWriterAgent: Takes a user specification and writes initial code
2. CodeReviewerAgent: Reviews the code and provides feedback
3. CodeRefactorerAgent: Refactors the code based on the review

The key element here is the SequentialAgent which orchestrates these agents, running them in order and passing state between them. The SequentialAgent "executes its sub_agents one after another in the order they are listed" and "passes the same InvocationContext sequentially, allowing agents to easily pass results via shared state."

CrewAI: A More Declarative Approach

While Google's ADK provides powerful orchestration capabilities, CrewAI offers a more declarative approach for building multi-agent systems that can automate repeatable, multi-step tasks. Its syntax is often considered more intuitive and readable.

Here's how to implement the same code pipeline using CrewAI:

from crewai import Crew, Process, Agent, Task, TaskOutput, CrewOutput

# Define your agents
code_writer = Agent(
  role='Code Writer',
  goal='Write clean, functioning Python code based on specifications',
  backstory='A senior Python developer with 10+ years of experience in writing efficient code',
  llm='gpt-4'
)

code_reviewer = Agent(
  role='Code Reviewer',
  goal='Thoroughly review code for bugs, style issues, and potential improvements',
  backstory='A detail-oriented code reviewer who specializes in Python best practices and PEP 8 standards',
  llm='gpt-4'
)

code_refactorer = Agent(
  role='Code Refactorer',
  goal='Implement reviewer suggestions and optimize code',
  backstory='An optimization expert who transforms code to be more efficient, readable, and maintainable',
  llm='gpt-4'
)

# Define your tasks
writing_task = Task(
  description='Write Python code that implements: {user_request}',
  agent=code_writer,
  expected_output='Clean Python code'
)

review_task = Task(
  description='Review the code for bugs and improvements: {code_from_writer}',
  agent=code_reviewer,
  expected_output='Code review comments',
  context=[writing_task]
)

refactor_task = Task(
  description='Refactor the code based on review feedback: {review_feedback}',
  agent=code_refactorer,
  expected_output='Optimized code',
  context=[writing_task, review_task]
)

# Form the crew with a sequential process
code_crew = Crew(
  agents=[code_writer, code_reviewer, code_refactorer],
  tasks=[writing_task, review_task, refactor_task],
  process=Process.sequential
)

# Execute the crew
result = code_crew.kickoff(inputs={"user_request": "Create a function to calculate factorial recursively"})

# Accessing the type-safe output
task_output: TaskOutput = result.tasks[0].output
crew_output: CrewOutput = result.output

The CrewAI implementation achieves the same result but with a more natural, human-readable structure. Instead of session state management, CrewAI uses a contextual approach where tasks can reference the outputs of previous tasks.

OpenAI Agents SDK: A Lightweight Approach

OpenAI recently released their Agents SDK, which provides a lightweight, production-ready framework for building agentic applications with minimal abstractions. Let's see how to implement our code pipeline example using OpenAI's Agents SDK:

from __future__ import annotations

import json
from agents import Agent, HandoffInputData, Runner, function_tool, handoff, trace
from agents.extensions import handoff_filters

# Define tools for our agents
@function_tool
def save_code(code: str) -> str:
    """Save the generated code to a file."""
    # In a real implementation, this would save to a file
    return "Code saved successfully."

@function_tool
def analyze_code(code: str) -> dict:
    """Analyze code for potential improvements."""
    # In a real implementation, this would do static analysis
    return {"issues": ["Consider adding error handling", "Add more comments"], "score": 8}

# Define our specialized agents
code_writer = Agent(
    name="Code Writer",
    instructions="""You are a Code Writer AI.
    Based on the user's request, write clean Python code.
    Be efficient and focus on best practices.""",
    tools=[save_code]
)

code_reviewer = Agent(
    name="Code Reviewer",
    instructions="""You are a Code Reviewer AI.
    Review the Python code provided and give constructive feedback.
    Focus on clarity, correctness, and efficiency.""",
    tools=[analyze_code],
    handoff_description="A specialized code reviewer that analyzes code quality and provides detailed feedback."
)

code_refactorer = Agent(
    name="Code Refactorer",
    instructions="""You are a Code Refactorer AI.
    Take the original code and review feedback, then refactor the code
    to address the feedback and improve its quality.""",
    tools=[save_code],
    handoffs=[handoff(code_reviewer)]
)

# Function to run our sequential pipeline
async def run_pipeline(user_request):
    # Trace to visualize the entire workflow
    with trace(workflow_name="Code Pipeline"):
        # Step 1: Generate code with the writer agent
        result = await Runner.run(
            code_writer,
            input=f"Create a Python function that: {user_request}"
        )

        generated_code = result.content

        # Step 2: Review the code with the reviewer agent
        result = await Runner.run(
            code_reviewer,
            input=result.to_input_list() +
                [{"role": "user", "content": f"Please review this code: {generated_code}"}]
        )

        review_feedback = result.content

        # Step 3: Refactor the code with the refactorer agent
        result = await Runner.run(
            code_refactorer,
            input=result.to_input_list() +
                [{"role": "user", "content": f"Refactor the code based on the review."}]
        )

        final_code = result.content

    return {
        "initial_code": generated_code,
        "review_comments": review_feedback,
        "final_code": final_code
    }

# Example usage
# await run_pipeline("calculate the factorial of a number recursively")

In this OpenAI Agents SDK implementation, we've defined three specialized agents and a function to orchestrate their sequential execution. Rather than using a built-in "sequential agent" primitive, the SDK takes a more explicit approach where developers manually handle the flow between agents using the conversation history.

Key Differences Between the Three Frameworks

Let's compare these three approaches to sequential multi-agent systems:

1. Architecture & Philosophy:

   • ADK uses a structured approach with specialized workflow agents like SequentialAgent
   • CrewAI uses a declarative, role-based approach with explicit Process.sequential
   • OpenAI Agents SDK uses a lightweight, minimal approach with manual orchestration

2. Syntax Style:

   • ADK uses a more programmatic approach with session state for sharing data
   • CrewAI uses a more declarative style with explicit task dependencies
   • OpenAI Agents SDK uses a conversation-based approach with message history

3. Agent Definition:

   • ADK focuses on instructions and descriptions
   • CrewAI adds concepts like roles, goals, and backstories
   • OpenAI Agents SDK focuses on minimalism with simple agent definitions

4. Data Flow:

   • ADK passes data via session state using output_key
   • CrewAI uses task contexts and dependencies
   • OpenAI Agents SDK passes data through conversation history and manual handling

5. Learning Curve:

   • ADK has a moderate learning curve with its state management approach
   • CrewAI is often considered the most intuitive for beginners
   • OpenAI Agents SDK is simple to start with but requires manual orchestration

6. Integration:

   • ADK is optimized for Google's ecosystem but supports third-party tools
   • CrewAI is more framework-agnostic
   • OpenAI Agents SDK is tightly integrated with OpenAI's models

Interestingly, these frameworks can work together in various ways. ADK provides a "CrewaiTool wrapper to integrate tools from the CrewAI library," and both ADK and CrewAI can be monitored with tools that also support the OpenAI Agents SDK.

Best Practices for Sequential Multi-Agent Systems

Regardless of which framework you choose, here are some best practices for building effective sequential multi-agent systems:

1. Clear Agent Specialization: Define each agent with a clear, focused role
2. Explicit Data Flow: Make data dependencies between agents explicit
3. Error Handling: Include mechanisms to handle failures at each step
4. Human-in-the-Loop: Consider adding human checkpoints for critical decisions
5. Testing: Test each agent individually before combining them
6. Monitoring: Implement logging to track the execution flow

Conclusion

Sequential multi-agent systems are really useful for tackling complex tasks by breaking them down into manageable steps handled by specialized agents. The three frameworks all offer robust ways for implementing these systems, each with its own strengths.

The choice between these frameworks ultimately depends on your specific needs, ecosystem preferences, and desired development style.

Start experimenting with these frameworks today to build sophisticated AI applications!

📫 DM Me for consulting inquiries and professional work.

Resources

• Google ADK Documentation
• CrewAI Documentation
• OpenAI Agents SDK Documentation