オーケストレーター・サブエージェントパターン

The Orchestrator-Subagent Mental Model

Responsibilities of the Orchestrator

from openai import OpenAI

client = OpenAI()

def orchestrator(goal: str) -> dict:
    # Step 1: Decompose
    plan = client.chat.completions.create(
        model='gpt-4o',
        messages=[
            {'role': 'system', 'content': 'You are a task planner. Decompose the goal into subtasks. Return JSON with keys: researcher_task, writer_task, coder_task.'},
            {'role': 'user', 'content': goal}
        ],
        response_format={'type': 'json_object'}
    )
    return plan.choices[0].message.content

Designing Specialized Subagents

def create_researcher_agent():
    return {
        'model': 'gpt-4o',
        'system': '''You are a research specialist. Your only job is to find accurate,
            cited information. Return structured findings with sources.
            Do NOT write prose or code - only research findings.''',
        'tools': ['search_web', 'fetch_url', 'query_arxiv'],  # 3 tools only
        'max_iterations': 10
    }

def create_writer_agent():
    return {
        'model': 'gpt-4o',
        'system': '''You are a technical writer. Transform research findings into clear,
            engaging prose. You receive structured data and return polished text.''',
        'tools': ['format_markdown', 'check_readability'],  # 2 tools only
        'max_iterations': 5
    }

The Delegation Protocol

from dataclasses import dataclass
from typing import Optional

@dataclass
class TaskPacket:
    task_id: str
    assignee: str          # which subagent receives this
    goal: str              # specific, actionable goal
    context: str           # only relevant background
    output_format: str     # JSON schema or description
    constraints: list[str] # e.g. ['max 500 words', 'cite sources']
    deadline_steps: int    # max iterations allowed

def delegate(packet: TaskPacket, subagent_fn) -> str:
    prompt = f'Goal: {packet.goal}\nContext: {packet.context}\nOutput format: {packet.output_format}\nConstraints: {packet.constraints}'
    return subagent_fn(prompt)

Collecting and Synthesizing Results

def synthesize_results(research: str, draft: str, code: str, goal: str) -> str:
    synthesis_prompt = f'''You are given outputs from three specialist agents.
    Your job is to synthesize them into one coherent final answer for this goal: {goal}

    RESEARCH FINDINGS:
    {research}

    WRITTEN DRAFT:
    {draft}

    CODE EXAMPLES:
    {code}

    Resolve any conflicts, fill gaps, and produce a unified final response.
    '''
    response = client.chat.completions.create(
        model='gpt-4o',
        messages=[{'role': 'user', 'content': synthesis_prompt}]
    )
    return response.choices[0].message.content

Sequential vs Parallel Delegation

import asyncio

# Sequential: writer needs research first
async def sequential_pipeline(goal):
    research = await researcher_agent(goal)
    draft = await writer_agent(research)  # depends on research
    return draft

# Parallel: all three tasks are independent
async def parallel_pipeline(topics):
    tasks = [
        researcher_agent(topics[0]),
        researcher_agent(topics[1]),
        researcher_agent(topics[2])
    ]
    results = await asyncio.gather(*tasks)  # runs simultaneously
    return orchestrator_synthesize(results)

Handling Subagent Failures

def delegate_with_retry(packet: TaskPacket, subagent_fn, max_retries=2):
    for attempt in range(max_retries + 1):
        try:
            result = subagent_fn(packet)
            validate_output(result, packet.output_format)
            return result
        except ValidationError as e:
            if attempt < max_retries:
                # Clarify the task for retry
                packet.goal += f'\n\nPrevious attempt failed: {str(e)}. Please fix.'
                print(f'Retry {attempt + 1} for task {packet.task_id}')
            else:
                return {'error': str(e), 'task_id': packet.task_id, 'status': 'failed'}

Prompting the Orchestrator Well

ORCHESTRATOR_SYSTEM_PROMPT = '''
You coordinate a team of specialist agents to complete complex tasks.

Available agents:
- researcher: Finds and cites factual information. Input: question string. Output: JSON with findings and sources.
- writer: Writes polished prose from structured data. Input: JSON findings. Output: markdown text.
- coder: Writes Python code. Input: natural language spec. Output: Python code string.

Workflow:
1. Analyze the goal and decide which agents are needed.
2. Identify dependencies: can any tasks run in parallel?
3. Delegate tasks with clear, specific goals.
4. Validate each result before proceeding.
5. Synthesize all results into a unified final answer.
'''

Context Isolation Is a Feature

Tracing and Observability

import time

def logged_delegate(packet: TaskPacket, subagent_fn):
    start = time.time()
    print(f'[DELEGATE] task_id={packet.task_id} assignee={packet.assignee}')
    print(f'[INPUT] {packet.goal[:100]}...')
    
    result = subagent_fn(packet)
    elapsed = time.time() - start
    
    print(f'[RESULT] task_id={packet.task_id} elapsed={elapsed:.2f}s')
    print(f'[OUTPUT] {str(result)[:100]}...')
    return result

Real-World Pattern: Report Generator

Quick Check

Lesson Recap