How to Orchestrate Multi-Agent Workflows with the OpenAI Agents SDK

You can orchestrate multi-agent workflows using three core patterns: parallel execution for concurrent tasks, agents-as-tools for sequential pipelines, and hybrid workflows that combine both approaches.

The OpenAI Agents SDK provides a lightweight, code-first framework for building multi-agent systems, and the Shubhamsaboo/awesome-llm-apps repository demonstrates production-ready patterns to orchestrate multi-agent workflows. By combining specialized agents through parallel execution, tool-based delegation, and hybrid orchestration, you can build robust AI pipelines that maximize quality, minimize latency, and maintain clear separation of concerns.

Core Patterns for Multi-Agent Orchestration

The SDK supports three fundamental design patterns for coordinating multiple agents, each optimized for different task structures and performance requirements.

Parallel Execution with Quality Selection

Parallel execution runs multiple independent agents simultaneously using asyncio.gather(), then aggregates or selects the best result. This pattern is ideal for tasks like translation quality-boosting, diverse content generation, or any scenario where agents operate independently and you want to optimize for the best output.

In ai_agent_framework_crash_course/openai_sdk_crash_course/9_multi_agent_orchestration/parallel_execution.py, the implementation creates a translator agent and a picker agent. Three translations are generated concurrently, then the picker selects the most natural version:

# Run three translations concurrently

res_1, res_2, res_3 = await asyncio.gather(
    Runner.run(spanish_agent, msg),
    Runner.run(spanish_agent, msg),
    Runner.run(spanish_agent, msg)
)

# Let the picker decide which is best

best_translation = await Runner.run(
    translation_picker,
    f"Original English: {msg}\n\nTranslations to choose from:\n{translations}"
)

All three calls share the same API quota and latency window because asyncio.gather schedules them simultaneously, dramatically cutting wall-clock time. The picker agent adds a quality-control layer, making the workflow robust to occasional noisy outputs.

Agents-as-Tools Orchestration

Agents-as-tools exposes a function tool wrapper around an agent so the orchestrator can call it like a normal function. The orchestrator decides the order and passes data between tools, creating structured pipelines such as research → write → edit where each step is handled by a dedicated expert agent.

In ai_agent_framework_crash_course/openai_sdk_crash_course/9_multi_agent_orchestration/agents_as_tools.py, each specialist is wrapped with @function_tool:

@function_tool
async def research_tool(topic: str) -> str:
    result = await Runner.run(
        research_agent,
        input=f"Research this topic thoroughly and provide key insights: {topic}",
        max_turns=3
    )
    return str(result.final_output)

# Orchestrator agent (high-level description)

content_orchestrator = Agent(
    name="Content Creation Orchestrator",
    instructions="""
    You coordinate research_tool → writing_tool → editing_tool to produce a polished article.
    """,
    tools=[research_tool, writing_tool, editing_tool]
)

When a user asks for an article, the orchestrator decides which tool to call first, passes the output to the next, and finally returns the edited result. This pattern makes the workflow declarative—the orchestrator's prompt encodes the workflow logic, while the SDK handles tool dispatch.

Hybrid Sequential and Parallel Workflows

Hybrid workflows mix the two approaches: stages are run sequentially, but inside a stage you can launch several agents in parallel. This is ideal for content-creation pipelines that need parallel research, then sequential synthesis, then parallel review.

The pattern is documented in the tutorial's README and implements a three-stage pipeline:

# Stage 1 – Parallel research from two agents

research_results = await asyncio.gather(
    research_agent_1.run(topic),
    research_agent_2.run(topic)
)

# Stage 2 – Sequential synthesis of research

combined = "\n".join([ItemHelpers.text_message_outputs(r.new_items) for r in research_results])
content = await writing_agent.run(combined)

# Stage 3 – Parallel review (quality, style)

reviews = await asyncio.gather(
    quality_agent.run(content),
    style_agent.run(content)
)

Key implementation ideas include breaking the problem into independent sub-tasks that can run in parallel, using sequential stages to aggregate results before moving on to preserve data dependencies, and adding fallback or validation agents in the final stage to ensure the output meets standards.

Complete Implementation Example

Below is a minimal, runnable script that demonstrates all three patterns. Save this as orchestrate_demo.py in the repository root and run it with python orchestrate_demo.py after installing dependencies from requirements.txt.

import asyncio
from agents import Agent, Runner, ItemHelpers, trace, function_tool

# ----- 1. Parallel translation with quality picker -----

spanish_agent = Agent(name="Spanish Translator",
                     instructions="Translate to fluent Spanish.")

picker = Agent(name="Translation Picker",
              instructions="""You are an expert translator. Choose the most natural
              Spanish version from the list and briefly explain why.""")

async def demo_parallel_translation():
    msg = "OpenAI agents let you build powerful AI workflows."
    with trace("Demo Parallel Translation"):
        r1, r2, r3 = await asyncio.gather(
            Runner.run(spanish_agent, msg),
            Runner.run(spanish_agent, msg),
            Runner.run(spanish_agent, msg),
        )
        translations = "\n\n".join(
            f"Version {i+1}: {ItemHelpers.text_message_outputs(res.new_items)}"
            for i, res in enumerate([r1, r2, r3])
        )
        best = await Runner.run(picker,
            f"Original: {msg}\n\nTranslations:\n{translations}")
    print("Best:", best.final_output)

# ----- 2. Agents-as-Tools content pipeline -----

research_agent = Agent(name="Researcher",
                      instructions="Provide a concise, factual overview of the topic.")
writing_agent = Agent(name="Writer",
                     instructions="Write a professional article based on given research.")
editing_agent = Agent(name="Editor",
                     instructions="Polish the article for grammar and flow.")

@function_tool
async def research_tool(topic: str) -> str:
    r = await Runner.run(research_agent,
                         input=f"Research: {topic}", max_turns=2)
    return str(r.final_output)

@function_tool
async def writing_tool(content: str) -> str:
    w = await Runner.run(writing_agent,
                         input=f"Write an article using this research: {content}",
                         max_turns=2)
    return str(w.final_output)

@function_tool
async def editing_tool(content: str) -> str:
    e = await Runner.run(editing_agent,
                         input=f"Edit and improve: {content}")
    return str(e.final_output)

orchestrator = Agent(
    name="Content Orchestrator",
    instructions="""
    Use research_tool → writing_tool → editing_tool to produce a polished article.
    """,
    tools=[research_tool, writing_tool, editing_tool]
)

async def demo_agents_as_tools():
    result = await Runner.run(
        orchestrator,
        "Create a 300‑word article on the future of AI‑assisted education."
    )
    print("\nFinal article:\n", result.final_output)

# ----- 3. Hybrid workflow (parallel research then sequential write) -----

async def demo_hybrid_workflow(topic: str):
    # Parallel research (two agents with different prompts)

    r1 = Runner.run(research_agent,
                    input=f"Technical aspects of {topic}", max_turns=2)
    r2 = Runner.run(research_agent,
                    input=f"Business impact of {topic}", max_turns=2)
    research_res = await asyncio.gather(r1, r2)

    combined = "\n\n".join(
        ItemHelpers.text_message_outputs(r.new_items) for r in research_res
    )

    # Sequential write & edit

    write_res = await Runner.run(writing_agent,
                                 input=f"Write a balanced article using:\n{combined}",
                                 max_turns=2)
    edit_res = await Runner.run(editing_agent,
                                input=f"Polish this article:\n{write_res.final_output}")

    print("\nHybrid result:\n", edit_res.final_output)

# ----- Run demos -----

async def main():
    await demo_parallel_translation()
    await demo_agents_as_tools()
    await demo_hybrid_workflow("generative AI in healthcare")

if __name__ == "__main__":
    asyncio.run(main())

This demo illustrates parallelism via asyncio.gather, tool wrapping via @function_tool, and hybrid flows that combine concurrent research with sequential synthesis.

Key Files and Resources

The following files in the Shubhamsaboo/awesome-llm-apps repository contain the reference implementations and documentation for these patterns:

File	Role	Path
README.md	Conceptual overview, pattern taxonomy, and learning objectives.	ai_agent_framework_crash_course/openai_sdk_crash_course/9_multi_agent_orchestration/README.md
parallel_execution.py	Minimal parallel-execution demo with quality-picker agent.	ai_agent_framework_crash_course/openai_sdk_crash_course/9_multi_agent_orchestration/parallel_execution.py
agents_as_tools.py	Full agents-as-tools orchestration example (research → write → edit) plus advanced conditional orchestrator.	ai_agent_framework_crash_course/openai_sdk_crash_course/9_multi_agent_orchestration/agents_as_tools.py
requirements.txt	Specifies the openai-agents SDK and its dependencies.	ai_agent_framework_crash_course/openai_sdk_crash_course/9_multi_agent_orchestration/requirements.txt
app.py	Optional Streamlit demo for interactive experimentation.	ai_agent_framework_crash_course/openai_sdk_crash_course/9_multi_agent_orchestration/app.py

Summary

To effectively orchestrate multi-agent workflows using the OpenAI Agents SDK, remember these key principles:

• Parallel Execution uses asyncio.gather() to run independent agents concurrently, reducing wall-clock time for tasks like translation or multi-perspective research.
• Agents-as-Tools wraps specialized agents with @function_tool, allowing an orchestrator agent to invoke them sequentially while maintaining clean separation of concerns.
• Hybrid Workflows combine parallel stages (for independent sub-tasks) with sequential stages (for dependent aggregation), optimizing both speed and coherence.
• Quality Control layers, such as picker agents or review agents, can be inserted at any stage to validate outputs and ensure robustness against individual agent failures.

Frequently Asked Questions

What is the difference between parallel execution and agents-as-tools?

Parallel execution runs multiple agents simultaneously using asyncio.gather() to speed up independent tasks, such as generating three translations at once. Agents-as-tools treats individual agents as callable functions via the @function_tool decorator, allowing an orchestrator agent to invoke them sequentially in a specific order, such as research followed by writing followed by editing.

How do I handle errors in multi-agent workflows?

You should wrap Runner.run() calls in try-except blocks to catch exceptions from individual agents, and design fallback agents that can retry or sanitize failed outputs. For critical workflows, implement validation agents that check outputs before passing them to the next stage, ensuring that errors are contained and do not propagate through the pipeline.

Can I combine these patterns with other orchestration frameworks?

Yes, the OpenAI Agents SDK patterns are compatible with broader orchestration frameworks like LangChain, LlamaIndex, or Prefect. You can use the SDK's Runner and Agent classes as specialized nodes within larger DAGs (Directed Acyclic Graphs), allowing you to leverage the SDK's lightweight agent execution while benefiting from external framework features like persistent state management or distributed tracing.

What are the performance benefits of parallel execution?

Parallel execution reduces total latency by utilizing asyncio.gather() to schedule multiple API calls simultaneously rather than waiting for each to complete sequentially. For workflows like multi-perspective research or translation quality boosting, this can reduce wall-clock time by 60-70% compared to sequential execution, though all calls still count against your API rate limits and token quotas.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}