다이내믹 워크플로우

다이내믹 워크플로우 (Dynamic Workflow)

DynamicWorkflow는 서브 에이전트 사이 의 조율이 실제 작업인 경우를 위한 것이에요. 몇 명의 전문가가 있다고 해 봐요. 한 명은 코드를 리뷰하고, 한 명은 결과를 요약하고, 한 명은 최종 메모를 써요. 각각은 단독으로 호출하기 쉽죠. 어려운 부분은 안무예요. 세 파일을 한 번에 리뷰하고, 무언가를 찾은 리포트만 유지하고, 그것들을 요약하고, 요약을 작가에게 넘기는 것. 팬아웃·체이닝·투표·재시도 루프가 개입된 조율이고, 그걸 한 모델 턴씩, 모든 중간 결과를 오케스트레이터 컨텍스트로 되돌려 흘려보내면서 실행하고 싶지 않을 때 이 capability를 잡으세요.

Pydantic AI Harness가 0.x 릴리스인 동안 API는 마이너 릴리스 사이에 바뀔 수 있어요. 바뀔 때는 폐기 경고와 릴리스 노트 마이그레이션 안내가 정확히 어떻게 업그레이드할지 알려줘요. 버전 정책 참고.

출처: 문서

본문

아이디어 (The idea)

서브 에이전트를 조율하는 평범한 방법은 스텝마다 도구 호출 하나를 하는 거예요. 에이전트가 리뷰어를 호출하고 기다리며, 결과를 읽고, 리뷰어를 다시 호출하고, 또 기다리고, 계속해요. 모든 중간 결과가 에이전트 컨텍스트로 돌아오고, 이전 것에 의존하는 모든 스텝은 별도의 모델 턴이에요.

DynamicWorkflow는 다른 길을 가요. 이름 붙은 서브 에이전트 카탈로그를 넘겨주면, 모델에게 run_workflow 도구 하나를 줘요. 그 도구 안에서 모델은 평범한 파이썬을 작성해요. 각 서브 에이전트가 호출·루프·조합할 수 있는 async 함수죠. 스크립트는 도구 호출 하나에서 끝까지 실행되고, 모델에게는 최종 값만 돌아와요. 안무가 대화에서 코드로 옮겨가요.

Code Mode를 만났다면 익숙할 거예요. 같은 Monty 샌드박스, 같은 아이디어. 많은 도구 호출 대신 스크립트 하나를 쓰는 것. 차이는 스크립트가 호출할 수 있는 것이에요. Code Mode에서는 에이전트 자신의 도구를 호출하고, 여기서는 전체 서브 에이전트를 호출해요.

Subagents와의 관계 (How this relates to Subagents)

하네스에는 두 개의 위임 capability가 있어요. 그것들은 같은 화폐 — 이름 붙고 격리된 서브 에이전트 실행 — 를 거래하지만 서로 다른 고도에서요:

SubAgents는 delegate_task(agent_name, task) 도구 하나를 노출해요. 각 위임은 자체 도구 호출이자 자체 모델 턴이에요. 위임이 드물거나, 각 결과가 다음 것 전에 부모의 판단이 필요할 때 맞는 선택이에요.
DynamicWorkflow는 안무를 스크립트로 옮겨요. 팬아웃·체이닝·투표·재시도 루프가 모두 한 도구 호출 안에서 실행되고, 중간 결과는 부모 컨텍스트에 들어오지 않아요.

확실하지 않으면 SubAgents부터 시작하세요. delegate_task 오케스트레이터는 서브 에이전트 자체를 바꾸지 않고 워크플로우 카탈로그로 변환돼요.

설치 (Installation)

스크립트는 Monty 샌드박스 안에서 실행되므로 엑스트라를 설치하세요:

pip install "pydantic-ai-harness[dynamic-workflow]"

uv add "pydantic-ai-harness[dynamic-workflow]"

첫 워크플로우 (Your first workflow)

서브 에이전트 둘, 오케스트레이터 하나:

from pydantic_ai import Agent
from pydantic_ai_harness import DynamicWorkflow

reviewer = Agent('openai:gpt-5', name='reviewer', description='Reviews code for bugs.')
summarizer = Agent('openai:gpt-5', name='summarizer', description='Summarizes findings.')

orchestrator = Agent(
    'openai:gpt-5',
    capabilities=[DynamicWorkflow(agents=[reviewer, summarizer])],
)

reviewer와 summarizer는 평범한 에이전트예요. 이미 아는 Agent 말이죠. 그들의 name이 모델이 스크립트에서 호출하는 함수 이름이 되므로, 유효한 파이썬 식별자인 이름을 고르세요. description은 모델에게 각각이 무엇인지 알려줘요. 함수를 문서화하듯 써요. DynamicWorkflow(agents=[...])가 그것들을 하나의 capability로 묶고 오케스트레이터에게 run_workflow 도구 하나를 줘요.

모델이 그것으로 하는 것 (What the model does with it)

오케스트레이터가 도구를 쓰기로 하면, 서브 에이전트를 하나씩 호출하지 않아요. 스크립트를 작성해요:

import asyncio

reports = await asyncio.gather(
    reviewer(task="Review auth.py for bugs:\n<file contents>"),
    reviewer(task="Review parser.py for bugs:\n<file contents>"),
)
await summarizer(task="Summarize these review findings:\n" + "\n\n".join(reports))

중요한 부분:

각 서브 에이전트는 async 함수예요. await로 호출해요.
작업을 단일 키워드 인자 task로 넘겨요. 항상 키워드로. reviewer(task="...")이지 reviewer("...")가 아니에요.
asyncio.gather(...)는 두 리뷰를 하나씩이 아니라 동시에 실행해요.
마지막 표현식의 값이 모델이 보는 결과가 돼요. 중간 reports 리스트는 샌드박스를 떠나지 않아요.

각 호출은 완전한 Agent.run이에요. 자체 모델 루프·메시지 히스토리·도구·타입화된 출력을 가져요. 두 가지가 따라와요. 호출은 격리되고(서브 에이전트는 이전 호출에서 아무것도 기억하지 않으므로, 필요한 모든 것을 task에 넣어요), 호출은 토큰이 들고 시간이 걸려요(그래서 이 capability가 아래에 예산을 주는 거예요).

서브 에이전트는 구조화 데이터를 반환할 수 있어요 (Sub-agents can return structured data)

서브 에이전트는 output_type이 만드는 무엇이든 반환해요. 기본은 문자열이지만, 서브 에이전트에 Pydantic 모델을 주면 스크립트가 dict를 받아요:

from pydantic import BaseModel

class Score(BaseModel):
    value: int
    reason: str

critic = Agent('openai:gpt-5', name='critic', description='Scores an answer 0-10.', output_type=Score)

스크립트 안에서 모델은 JSON 객체를 읽듯 서브스크립트로 필드를 읽어요:

result = await critic(task="Score this answer: ...")
result["value"]   # not result.value

모델이 보는 카탈로그는 각 출력 타입을 TypedDict로 렌더링하므로, 필드를 알고 스스로 서브스크립트로 읽어요.

결과가 돌아오는 방식 (How results come back)

스크립트 마지막 표현식의 값이 도구 결과가 돼요. 모델이 print()하지 않아요.

스크립트가...	모델이 받는 것
print 없이 값으로 끝남	그 값 직접 (`None`이면 `{}`)
print하고 값으로 끝남	`{"output": "<printed text>", "result": <value>}`
print하고 `None`으로 끝남	`{"output": "<printed text>"}`

print()는 디버그 로깅용이에요. 문자열화하므로, 실제 결과는 마지막 표현식이 나르게 하세요.

서브 에이전트 모델 고르기 (Choosing sub-agent models)

기본적으로 각 서브 에이전트는 구성된 모델을 사용해요. 호스트가 부모 에이전트에 실행별 모델 오버라이드(예: /model 명령에서)를 넘기고 모든 서브 에이전트 디스패치가 그 해석된 부모 모델을 따라야 하면 inherit_model=True를 설정하세요. 서브 에이전트를 의도적으로 다른 모델에 고정하면 False로 남겨 두세요.

안전하게 유지하기: 예산 (Keeping it safe: budgets)

서브 에이전트는 비결정적이고, 토큰이 들며, 더 많은 서브 에이전트로 팬아웃할 수 있어요. DynamicWorkflow는 하드한 횟수 상한, 토큰 예산, 그리고 폭주하는 샌드박스 스크립트에 대한 가드를 줘요.

`max_agent_calls` — 정확한 횟수

DynamicWorkflow(agents=[...], max_agent_calls=50)  # 50 is the default

한 부모 실행에서 서브 에이전트 실행 수에 대한, 호스트가 강제하는 하드 상한이에요. 그 실행의 모든 run_workflow 호출에서 공유되는 하나의 예산이며, 스크립트가 asyncio.gather로 팬아웃해도 정확히 유지돼요. 예산이 소진되면 워크플로우는 서브 에이전트 호출을 멈추고, 최근 완료된 결과 최대 20개의 경계 있는 미리보기와 함께 종료 결과를 반환해요. 이것이 실행 수를 정확히 경계 짓는 유일한 손잡이예요.

`sub_agent_usage_limits`와 `forward_usage` — 비용 경계짓기

sub_agent_usage_limits는 각 서브 에이전트 실행에 적용되는 UsageLimits예요. forward_usage는 전체 트리가 하나의 사용량 카운터를 공유할지 제어해요:

`forward_usage`	카운터	한도가 뜻하는 것
`True` (기본)	부모의 `usage`를 트리 전체가 공유	트리 전체 상한. 동시 팬아웃 아래서는 best-effort: 여러 서브 에이전트가 카운트에 더하기 전에 검사를 통과할 수 있음
`False`	각 서브 에이전트 실행이 자체적으로 카운트	실행별 한도. `T`의 per-run `total_tokens_limit`에 `N`의 `max_agent_calls`는 트리를 대략 `N * T` 토큰으로 경계 지음

부모 run() 사용량 한도는 전달되지 않아요

부모 run()에 넘기는 usage_limits는 서브 에이전트로 전달되지 않아요. 부모 자신의 요청 경계에서만 다시 검사돼요. 서브 에이전트를 경계 짓으려면 sub_agent_usage_limits를, 실행 수에 대한 정확한 상한은 max_agent_calls를 쓰세요.

`resource_limits` — 스크립트 자체 가드

이 한도들은 오케스트레이션 스크립트 자신의 메모리를 가드해요. 호출하는 서브 에이전트는 아니에요. 기본 백스톱은 시간 제한 없는 256MB예요. 인쇄된 출력은 Monty의 10 MiB 기본 상한과 함께 별도로 수집돼요.

DynamicWorkflow(agents=[...], resource_limits={'max_duration_secs': 30})

max_duration_secs는 스크립트가 샌드박스 코드를 실행하는 데 쓰는 시간을 측정하지, 벽시계 시간을 측정하지 않아요. 스크립트가 서브 에이전트를 기다리는 동안은 중단되어 그 시간은 세지 않으므로, 서브 에이전트가 아무리 오래 걸려도 정상 워크플로우에서는 상한이 발동하지 않아요. 그것의 유일한 역할은 순수 CPU 폭주 — await하지 않는 while True: 루프 — 를 잡는 것이고, 서브 에이전트 예산 중 어느 것도 서브 에이전트를 호출하지 않으므로 멈출 수 없어요. 'unlimited'를 넘기면 모든 샌드박스 리소스 한도를 제거해요. 별도의 10 MiB print 상한은 남아요. 부분 dict는 백스톱에 병합되어 이름 붙인 상한만 오버라이드해요.

워크플로우는 중첩되지 않아요 (Workflows do not nest)

서브 에이전트는 자기 워크플로우를 시작할 수 없어요. 중첩된 run_workflow 호출은 실행 대신 종료 오류를 반환해요. 실용적 규칙: 카탈로그의 서브 에이전트에게 DynamicWorkflow capability를 주지 마세요. 그것들은 오케스트레이션의 잎이지 오케스트레이터가 아니에요.

서브 에이전트 이름 바꾸기: `WorkflowAgent`

기본적으로 서브 에이전트는 자기 name과 description 아래 나타나요. 에이전트 자체를 편집하지 않고 한 워크플로우에 다른 이름이나 설명을 주려면 WorkflowAgent로 감싸세요:

from pydantic_ai_harness.dynamic_workflow import WorkflowAgent

DynamicWorkflow(
    agents=[
        WorkflowAgent(
            reviewer,
            name='check',
            description='Checks one code change and returns actionable review findings.',
        ),
    ],
)

이제 모델은 check(task=...)를 호출해요. 베어 에이전트를 넘기는 것은 오버라이드 없이 WorkflowAgent로 감싸는 축약이에요.

실행 중 서브 에이전트 추가: `reveal()`

카탈로그는 시작 시점에 고정돼, 턴 사이에 프롬프트-캐시 프리픽스에 유지돼요. 실행 중에 새 서브 에이전트를 가능하게 하려면(예: fixer 에이전트가 프로비저닝된 뒤), DynamicWorkflow 인스턴스의 참조를 유지하고 reveal()을 호출하세요:

workflow = DynamicWorkflow(agents=[reviewer])
orchestrator = Agent('openai:gpt-5', deps_type=MyDeps, capabilities=[workflow])

# later, from the host or from another tool:
workflow.reveal(fixer)

reveal된 서브 에이전트는 다음 스텝에서 호출 가능해지고, 모델은 새 함수의 시그니처를 담은 짧은 공지 메시지로 그것을 알게 돼요. run_workflow 설명 자체는 실행 시작 시 존재한 에이전트에 고정된 채로 있어, 런타임 reveal이 프롬프트-캐시 프리픽스를 절대 옮기지 않아요. reveal()은 추가 전용이고 즉시 검증해요. 누락된 이름·잘못된 식별자·예약 키워드·이름 충돌은 호출 지점에서 UserError를 발생시켜요.

필요할 때만 로드: `defer_loading`

DynamicWorkflow는 꽤 많은 지시문 텍스트를 나르고, 대부분의 턴은 그것을 필요로 하지 않아요. 모델이 실제로 로드할 때까지 한 줄 항목으로 접어 두세요:

DynamicWorkflow(
    agents=[reviewer, summarizer],
    id='workflow',
    defer_loading=True,
)

defer_loading=True는 안정적인 id가 필요해요. 전체 그림은 on-demand capabilities를 참고하세요.

샌드박스에서 실행되는 것 (What runs in the sandbox)

스크립트는 파이썬의 부분집합인 Monty에서 실행돼요. 경계를 아는 것이 중요해요:

서드파티 라이브러리 없음.
임포트 가능한 표준 라이브러리 모듈: sys, typing, asyncio, math, json, re, unicodedata, datetime, os, pathlib. 쓰는 것을 임포트해요. 파일시스템·환경·시계 연산은 워크플로우 스크립트에 대해 구성되지 않아요.
벽시계나 타이밍 프리미티브 없음 — asyncio.sleep 없음, datetime.datetime.now() 없음, datetime.date.today() 없음, time 모듈 없음.
asyncio.gather(...)는 위치 인자의 awaitables로 서브 에이전트를 동시에 실행하고 키워드 인자는 없어요(return_exceptions=True 포함). 다른 작업 생성·대기 API는 사용 불가.

스크립트가 실행되기 전에 서브 에이전트 시그니처에 대해 정적으로 타입 검사돼요. 철자가 틀린 함수, 위치 task, 잘못된 타입 인자 같은 평범하고 정적으로 증명 가능한 실수는 한 번의 재시도가 들지, 서브 에이전트 예산이나 샌드박스 실행은 들지 않아요. Any로 타입된 값은 런타임 검증에 닿을 수 있고, 서브 에이전트가 실행되기 전에 여전히 거부돼요.

잡지 않은 오류는 전체 스크립트를 중단해요

서브 에이전트 실패는 RuntimeError로 표면화돼요. 스크립트는 try/except RuntimeError로 잡을 수 있고, 잡지 않은 실패는 전체 스크립트를 중단하고 모델이 재시도해요. 스크립트가 일부 서브 에이전트가 이미 끝난 뒤 실패하면, 재시도 프롬프트는 최근 완료된 결과 최대 20개의 경계 있는 미리보기를 나열해요. 모델은 잘리지 않은 미리보기를 평범한 값으로 재사용해 같은 호출에 다시 비용을 지불하지 않을 수 있어요.

관측 가능성 (Observability)

Logfire 트레이스는 워크플로우가 무엇을 했는지 보는 최고의 방법이에요. 각 서브 에이전트 실행이 run_workflow 스팬 아래 중첩되어 나타나고, 스팬은 모델이 쓴 정확한 code 인자를 담아 실제로 실행한 스크립트를 읽을 수 있어요. 일급 진행 스트리밍이 배송될 때까지, 각 서브 에이전트 Agent에 event_stream_handler를 설정해 한 도구 호출 안에서 서브 에이전트 실행을 지켜보세요.

API

DynamicWorkflow(                  # all parameters are keyword-only
    agents=[...],                 # Sequence[AbstractAgent | WorkflowAgent], required
    tool_name='run_workflow',
    max_agent_calls=50,
    max_retries=3,
    forward_usage=True,
    inherit_model=False,          # True -> sub-agents run with the parent run's resolved model
    sub_agent_usage_limits=None,  # UsageLimits per sub-agent run; None -> pydantic-ai default
    resource_limits=None,         # None -> backstop (256 MB, no time cap);
                                  # 'unlimited' -> sandbox limits off; a dict merges onto the backstop
    id=None,                      # required when defer_loading=True
    description=None,             # one-line catalog entry shown while deferred
    defer_loading=False,
)

workflow.reveal(agent)            # AbstractAgent | WorkflowAgent; validates before appending

WorkflowAgent(
    agent,                        # AbstractAgent, required, positional
    name=None,                    # sandbox function name; falls back to agent.name
    description=None,             # function docstring; falls back to agent.description
)

DynamicWorkflowToolset과 WorkflowResourceLimits도 고급 사용을 위해 모듈에서 내보내져요.

소스: pydantic_ai_harness/dynamic_workflow/.

더 읽기 (Further reading)

Code Mode — 같은 샌드박스, 서브 에이전트 대신 에이전트 자신의 도구 호출.
Subagents — 스크립트된 안무 없이, 도구 호출당 위임 하나인 서브 에이전트.
Rewriting Bun in Rust (Bun) — Claude Code의 다이내믹 워크플로우를 통한, 규모에서의 같은 패턴.
Capabilities와 on-demand capabilities.

더 알아보기 (Learn more)

Code Mode — 같은 Monty 샌드박스.
Subagents — 도구 호출당 위임 하나.
Pydantic AI Harness — 패키지 전반.

다이내믹 워크플로우