Planning

Planning

이 문서에서는 Planning capability를 소개해요. 작은 도구 모음을 통해 모델에 구조화되고 스스로 갱신되는 작업 목록을 제공하며, 매 턴마다 현재 계획을 프롬프트 캐시를 무효화하지 않고 모델에 다시 노출해요. 단일 실행 동안 메모리에 유지하거나 SQLite/Postgres에 영속할 수 있고, 의존성을 가진 하위 작업으로 단계를 나누고, 세밀한 변경에서 이벤트를 발생시킬 수 있어요.

출처: 문서

본문

Planning은 작은 도구 모음을 통해 모델에 구조화되고 스스로 갱신되는 작업 목록을 제공하고, 매 턴마다 현재 계획을 프롬프트 캐시를 절대 무효화하지 않고 모델에 다시 노출해요. 단일 실행 동안 메모리에 유지하거나 SQLite/Postgres에 영속할 수 있고, 의존성을 가진 하위 작업으로 단계를 나누며, 세밀한 변경에서 이벤트를 발생시킬 수 있어요.

소스

이 capability는 독립형 pydantic-ai-todo 라이브러리의 작업 목록 기능(영속 저장소, 하위 작업, 의존성, 이벤트)을 통합하며, 그 라이브러리를 대체해요. pydantic-ai-todo에서 마이그레이션한다면 도구 이름이 바뀌었어요:

pydantic-ai-todo

Planning

write_todos

write_plan

read_todos

read_plan

add_todo

add_task

update_todo_status / update_todo_statuses

update_task_status / update_task_statuses

remove_todo

remove_task

add_subtask, set_dependency, get_available_tasks

변경 없음

계획해야 할 두 가지 차이점이 있어요. connection-string 편의 기능이 없고(create_storage(backend=...) 등이 사라졌어요 — 직접 asyncpg 풀이나 Redis 클라이언트를 구성해야 하며, 이것이 harness를 드라이버 없이 유지하게 해요), PlanEventtimestamp를 지니지 않으므로 이를 기준으로 정렬하거나 기록하는 소비자는 자체 시계를 공급해야 해요.

Pydantic AI Harness가 0.x 릴리스인 동안에는 minor 릴리스 사이에 API가 바뀔 수 있어요. 그럴 때는 deprecation 경고와 릴리스 노트 마이그레이션 안내가 (당신이나 당신의 에이전트에게) 정확히 업그레이드 방법을 알려줘요. 버전 정책을 참고하세요.

The problem

긴 에이전트 실행은 표류하기 마련이에요. 모델이 원래 무엇을 하려 했고 무엇이 남았는지 잃어버리게 되죠. 흔한 해결책(진행 중인 계획을 유지하고 매 턴 시스템 프롬프트에 다시 주입)은 프롬프트 캐시를 무효화해요. 시스템 프롬프트는 요청의 맨 앞에 있으므로, 계획을 편집할 때마다 캐시된 접두사가 바뀌고 전체 대화를 전체 토큰 비용으로 다시 처리해야 해요.

The solution

모델은 planning 도구 모음을 통해 계획을 소유해요. 현재 계획은 각 요청의 꼬리에 붙는 임시(ephemeral) 리마인더로 다시 노출되며, 단일 캐시 breakpoint는 마지막 영속 사용자 콘텐츠에 앵커링돼요:

  • 리마인더는 영속 히스토리가 저장된 뒤 추가되므로 모델에는 도달하지만 message_history에는 절대 기록되지 않아요. 턴마다 리마인더가 누적되지 않아요.
  • CachePoint는 마지막 영속 사용자 콘텐츠에 위치하므로, 그것이 저장하는 접두사는 다음 요청의 접두사가 되고 캐시 히트가 턴에서 턴으로 유지돼요. 리마인더는 breakpoint를 지니지 않으므로 가변 콘텐츠를 재전송해도 캐시를 무효화하지 않아요.

모든 capability 캐시 breakpoint와 마찬가지로 프로바이더 매핑이 적용돼요. OpenAI 모델은 모델 프로필이 명시적 캐시 제어를 활성화할 때만 CachePoint를 받고, 앵커할 영속 사용자 콘텐츠가 없으면 리마인더가 breakpoint 없이 전송돼요.

앵커는 요청에 존재하는 마지막 UserPromptPart에 위치해요. Planning보다 앞에 나열된 capability가 매 요청마다 사용자 콘텐츠를 추가한다면(예: SystemReminders) 앵커를 해당 파트로 옮기므로, 접두사는 콘텐츠가 턴 사이에 안정적인 동안에만 캐시 안정적이에요.

Usage

Planning()capabilities에 넣어 Agent를 구성하세요. 도구는 자동으로 등록되고 정적 사용 안내가 시스템 프롬프트에 추가돼요:

from pydantic_ai import Agent
from pydantic_ai_harness import Planning

agent = Agent('anthropic:claude-sonnet-4-6', capabilities=[Planning()])

result = agent.run_sync('Refactor the auth module and add tests.')
print(result.output)

The tools

Tool

Purpose

write_plan(items)

전체 계획을 생성하거나 교체해요(전체 목록 교체).

read_plan()

단계 id와 진행 요약을 포함해 현재 계획을 읽어요.

add_task(content, active_form)

단일 pending 단계를 추가해요.

update_task_status(task_id, status)

한 단계를 id로 상태 사이에서 이동해요.

update_task_statuses(updates)

여러 상태 변경을 한 번의 호출로 적용하며, 전부 아니면 전무(all-or-nothing)로 검증해요.

remove_task(task_id)

한 단계를 id로 삭제해요.

각 단계는 content 문자열, 선택적 현재진행형 active_form 라벨, 그리고 status(pending, in_progress, completed, cancelled)로 구성돼요. 관례(안내와 도구 응답에 명시됨)는 정확히 하나의 단계를 in_progress로 유지하는 것이에요.

여섯 개 모두 기본적으로 등록돼요. tools=는 이를 allowlist로 좁히며, 내장 안내가 그에 따라 동작해요:

from pydantic_ai_harness import Planning

planning = Planning(tools=['write_plan'])  # 전체 계획 교체만 -- 도구 하나, 추적할 단계 id 없음

현재 모드가 등록하지 않는 도구 이름을 지정하면 ValueError가 발생하며, descriptions의 알 수 없는 키도 마찬가지예요.

Subtasks and dependencies

enable_subtasks=True를 전달하면 도구 세 개, blocked 상태, 그리고 read_planhierarchical 뷰가 추가돼요:

Tool

Purpose

add_subtask(parent_id, content, active_form)

부모 아래에 자식 단계를 추가해요.

set_dependency(task_id, depends_on_id)

한 단계가 다른 단계를 기다리게 해요. 의존 단계는 사전 조건이 해결될 때(completed 또는 cancelled)까지 자동으로 blocked돼요. 자기 의존, 순환, 중복은 거부돼요.

get_available_tasks()

불완전한 의존성이 없는 단계(지금 시작할 수 있는 단계)를 나열해요.

parent_id, depends_on, blocked 상태는 enable_subtasks가 설정되지 않으면 write_plan이 거부해요. 하위 작업 도구가 없으면 의존성을 조정하는 것이 없고 계층을 렌더링하는 뷰도 없으므로, 저장하는 것이 계획이 반영하지 않는 쓰기가 되기 때문이에요.

Persistence

기본적으로 계획은 실행마다 새롭고 고립된 인메모리 계획이에요. store를 전달하면 영속시킬 수 있어요:

from pydantic_ai_harness import Planning
from pydantic_ai_harness.planning import SqlitePlanStore

planning = Planning(store=SqlitePlanStore('plan.db', session='user-123'))

내장 저장소는 InMemoryPlanStore, SqlitePlanStore, PostgresPlanStore(호출자가 소유한 asyncpg 풀 위), RedisPlanStore(호출자가 소유한 redis.asyncio 클라이언트 위)예요 — 그래서 harness는 데이터베이스 드라이버가 필요 없어요. 어떤 PlanStore 구현도 동작하며, store_resolver가 실행마다 하나를 선택해요. SqlitePlanStore는 파일 기반 데이터베이스가 필요해요. 임시 계획에는 ':memory:'보다 InMemoryPlanStore를 사용하세요.

꼬리 리마인더는 모델 요청마다 저장소를 읽으므로, 저장소가 예외를 발생시키면 성능 저하 대신 실행이 실패해요 — 리마인더는 best-effort가 아니에요. 이는 의도적이에요. 모델이 더 이상 볼 수 없는 계획은 조용히 계속 실행할 상태가 아니기 때문이에요. 재시도와 폴백 정책은 Planning이 아니라 저장소의 몫이며, PlanStore는 정확히 그것을 감쌀 수 있도록 protocol로 정의되어 있어요:

class BestEffort:
    """Serve the last known plan when the backing store is unreachable."""

    def __init__(self, inner: PlanStore) -> None:
        self._inner, self._last = inner, []

    async def get_items(self) -> list[PlanItem]:
        try:
            self._last = await self._inner.get_items()
        except ConnectionError:
            pass
        return self._last

    # ... delegate the other five methods to `self._inner`

Planning and executing in separate runs

공유 저장소는 두 실행 사이의 전체 핸드오프 메커니즘이에요. 한 에이전트가 계획을 쓰고 두 번째 에이전트가 실행하며, 계획이 둘 사이를 가로지르는 유일한 상태예요:

store = SqlitePlanStore('plan.db', session='issue-403')

planner = Agent('anthropic:claude-opus-4-7', capabilities=[Planning(store=store)])
executor = Agent('anthropic:claude-sonnet-4-6', capabilities=[Planning(store=store)])

await planner.run('Investigate the issue and write a plan. Do not implement anything.')
await executor.run('Implement the plan.')

실행자는 message_history 없이 시작하므로 플래너의 조사에 대한 비용을 절대 지불하지 않아요. 첫 요청은 새 프롬프트와 계획 리마인더만 지니며, 리마인더는 capability가 저장소에서 재구성해요. 그래서 두 에이전트가 다른 모델에서 실행될 수 있어요. 대형 컨텍스트 모델이 읽기와 추론을 하고, 더 작은 모델이 결과 체크리스트에 대해 실행할 수 있죠.

플래너의 읽기 전용 규율은 그 에이전트를 어떻게 구성하는지(어떤 툴셋을 받고 지침이 무엇을 말하는지)의 속성이지, capability가 강제하는 것이 아니에요.

Events

타입이 지정된 계획 이벤트를 구독해 Planning 도구를 통해 이루어진 변경에 반응할 수 있어요:

from pydantic_ai import Agent
from pydantic_ai_harness import Planning
from pydantic_ai_harness.planning import PlanCompletedEvent

agent = Agent('anthropic:claude-sonnet-4-6', capabilities=[Planning()])

@agent.on_event(PlanCompletedEvent)
async def announce(ctx, event):
    print('done:', event.item.content)

계열에는 PlanCreatedEvent, PlanUpdatedEvent, PlanStatusChangedEvent, PlanCompletedEvent, PlanDeletedEvent가 있고, 각각 영향을 받은 item을 지니며, 업데이트에는 previous_state도 지녀요.

실행 이벤트는 write_plan을 포함한 planning 도구 경로에서 나와요. PlanStore에 대한 직접 애플리케이션 변경은 실행 컨텍스트가 없으므로 실행 이벤트를 만들지 않아요. PlanEventEmitter, EventCallback, 저장소 event_emitter 파라미터는 계속 지원되지만 deprecate됐어요.

Why whole-plan replacement

가변 정수 인덱스(insert/remove/reorder)로 단계를 다루는 것은 코드와 모델 모두에게 오류가 쉽기 쉬워요. write_plan은 매 호출마다 전체 계획을 재진술하므로 추적할 인덱스가 없어요. 세밀한 편집(add_task, update_task_status, remove_task)은 대신 read_plan이 보여주는 안정적인 id를 참조해요.

Caching guarantee

계획은 시스템 프롬프트나 instructions에 절대 주입되지 않아요. 정적 사용 안내는 거기(캐시 안정)로 가고, 가변 계획만 임시 꼬리 리마인더에 실리며, 리마인더는 요청별 복사본에만 존재하고 절대 영속되지 않아요. inject=False로 비활성화할 수 있어요. Pydantic AI는 프롬프트 캐싱을 지원하는 프로필을 가진 모델에 CachePoint를 매핑하고, 다른 모델에서는 무시돼요.

영속 실행 capability가 붙어 있으면 리마인더를 만드는 데 쓰는 계획 읽기는 저널링된 capability 연산이에요. 재생은 저장소를 다시 읽는 대신 기록된 계획을 재사용해요. Planning은 안정적인 기본 id='planning'을 지니므로 영속 복구가 구성 없이 동작해요.

Configuration

from pydantic_ai_harness import Planning

Planning(
    guidance=None,           # static system-prompt guidance; None = default, '' = omit
    cache_ttl='5m',          # TTL for the cache breakpoint anchored on the last durable user content ('5m' | '1h')
    store=None,              # None = fresh in-memory plan per run; or a PlanStore to persist
    enable_subtasks=False,   # add subtask/dependency tools and the 'blocked' status
    inject=True,             # surface the current plan as a cache-safe tail reminder
    tools=None,              # None = every tool the mode registers; or an allowlist of names
    descriptions=None,       # optional per-tool description overrides, keyed by tool name
)

Agent spec (YAML/JSON)

Planning은 Pydantic AI의 agent spec과 함께 동작해요:

# agent.yaml
model: anthropic:claude-sonnet-4-6
capabilities:
  - Planning: {}
from pydantic_ai import Agent
from pydantic_ai_harness import Planning

agent = Agent.from_file('agent.yaml', custom_capability_types=[Planning])
result = agent.run_sync('...')
print(result.output)

Further reading

API reference

Planning

Bases: AbstractCapability[AgentDepsT]

프롬프트 캐시를 절대 무효화하지 않는 구조화된 작업 계획.

모델은 작은 도구 모음(write_plan, read_plan, add_task, update_task_status, update_task_statuses, remove_task, 그리고 enable_subtasks가 설정되면 add_subtask, set_dependency, get_available_tasks)을 통해 계획을 소유해요. tools는 그 표면을 allowlist로 좁혀요. 현재 계획은 각 요청의 꼬리에 붙는 임시 리마인더로 다시 노출돼요. 단일 CachePoint가 마지막 영속 사용자 콘텐츠에 앵커링되므로, 그것이 저장하는 접두사는 다음 요청의 접두사가 되고, 리마인더 자체는 breakpoint를 지니지 않으므로 매 턴 가변 계획 콘텐츠만 다시 읽혀요.

기본적으로 계획은 단일 실행 기간 동안 메모리에 저장돼요(실행마다 새롭고 고립된 계획). store(또는 store_resolver)를 전달해 영속시킬 수 있어요 — 예: SqlitePlanStore 또는 PostgresPlanStore.

from pydantic_ai import Agent
from pydantic_ai_harness.planning import Planning

agent = Agent('anthropic:claude-sonnet-4-6', capabilities=[Planning()])

Attributes

guidance

시스템 프롬프트용 정적 계획 안내. 캐시 안정.

세 가지 상태이므로, 옵트아웃은 우연이 아니라 의도적으로 하는 것이에요:

  • None(기본값): 내장 안내를 사용.
  • '': 안내 전혀 없음.
  • 기타 문자열: 내장 안내 대신 그 값을 사용.

None이 "안내 없음"을 의미하는 단일 str | None은 기본값을 명시적으로 요청할 방법이 없고, None으로 해석되는 구성을 조용한 옵트아웃으로 만들어요. 이는 memory, exa, runtime_authoring이 같은 방식으로 읽히는 것과 일치해요.

Type: str | None Default: None

cache_ttl

마지막 영속 사용자 콘텐츠에 앵커링된 캐시 breakpoint의 TTL.

Type: Literal['5m', '1h'] Default: '5m'

store

저장 백엔드. None은 실행마다 새 인메모리 계획을 유지(원래의 임시 동작)해요. 실행 간에 계획을 영속하려면 store를 전달하세요.

Type: PlanStore | None Default: None

store_resolver

선택적 실행별 store 리졸버, 예: lambda ctx: ctx.deps.plan_store.

Type: Callable[[RunContext[AgentDepsT]], PlanStore] | None Default: None

enable_subtasks

true일 때 하위 작업/의존성 도구와 blocked 상태를 추가.

Type: bool Default: False

inject

매 턴 현재 계획을 임시 꼬리 리마인더로 노출.

Type: bool Default: True

tools

등록할 도구 이름의 선택적 allowlist. None은 모두 등록.

전체 표면은 write_plan, read_plan, add_task, update_task_status, update_task_statuses, remove_task이며, enable_subtasks 아래에는 add_subtask, set_dependency, get_available_tasks가 있어요. tools=['write_plan']은 가장 작은 유용한 계획 표면이에요. 이 모드가 등록하지 않는 도구 이름을 지정하면 ValueError가 발생해요.

내장 guidance는 전체-계획/세밀/하위 작업 구분에 대해 allowlist를 따르고, 그룹 내 절단은 자체 guidance 문자열과 짝지을 때 더 낫습니다.

Type: Sequence[str] | None Default: None

descriptions

도구 이름 키로 된 선택적 도구별 설명 재정의. 알 수 없는 이름은 ValueError를 발생시켜요.

Type: dict[str, str] | None Default: None

Methods

for_run

@async

def for_run(ctx: RunContext[AgentDepsT]) -> Planning[AgentDepsT]

이 실행의 store를 해석하고 캐시한 복제본을 반환해요(실행별 고립).

Returns

Planning[AgentDepsT]

resolve_store
def resolve_store(ctx: RunContext[AgentDepsT]) -> PlanStore

캐시된 실행 store를 반환하거나, 직접 툴셋 사용을 위해 하나를 해석해요.

Returns

PlanStore

get_toolset
def get_toolset() -> AgentToolset[AgentDepsT] | None

이 실행의 해석된 store 위에 planning 툴셋을 제공해요.

Returns

AgentToolset[AgentDepsT] | None

get_instructions
def get_instructions() -> AgentInstructions[AgentDepsT] | None

planning 도구 사용에 대한 정적이고 캐시 안정적인 안내를 제공해요.

커스텀 guidance 문자열은 그대로 사용돼요. 기본값은 실제로 등록된 도구에서 조립돼요 — tools가 granular 도구를 모두 제외하면 그 문장이 빠지고, enable_subtasks 아래에서 하위 작업/의존성 워크플로가 추가되므로 — 모델에게 없는 도구에 대해 말하지 않아요.

Returns

AgentInstructions[AgentDepsT] | None

wrap_model_request

@async

def wrap_model_request(
    ctx: RunContext[AgentDepsT],
    *,
    request_context: ModelRequestContext,
    handler: WrapModelRequestHandler,
) -> ModelResponse

영속 사용자 콘텐츠에 캐시 breakpoint를 앵커링한 다음 임시 계획 리마인더를 추가해요.

Returns

ModelResponse

from_spec

@classmethod

def from_spec(
    cls,
    *,
    backend: Literal['memory', 'sqlite'] = 'memory',
    database: str = '.agent-plan.db',
    session: str = 'default',
    enable_subtasks: bool = False,
    inject: bool = True,
    guidance: str | None = None,
    cache_ttl: Literal['5m', '1h'] = '5m',
    tools: list[str] | None = None,
) -> Planning[AgentDepsT]

직렬화 가능한 옵션에서 Planning capability를 구성해요.

Returns

Planning[AgentDepsT]

get_serialization_name

@classmethod

def get_serialization_name(cls) -> str | None

agent-spec 지원용 직렬화 이름.

Returns

str | None

더 알아보기 (Learn more)