오케스트레이터-하위 에이전트 패턴

Orchestrator-하위 에이전트 사고 모델

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

전문화된 하위 에이전트 설계

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
    }

위임 프로토콜

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)

결과 수집 및 종합

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

순차 위임과 병렬 위임 비교

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)

서브에이전트 오류 처리

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'}

오케스트레이터를 효과적으로 프롬프트하기

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.
'''

컨텍스트 격리는 기능입니다

추적 및 관찰 가능성

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

실제 패턴: 보고서 생성기

빠른 확인

레슨 요약