Deterministic 워크플로우

호출 순서가 업무 계약의 일부라면 Deterministic workflow를 사용하세요. controller LLM이 없으며 run_turn(ctx)에 적힌 순서만 실행합니다. 정산, 여러 시스템의 병렬 조회, 애플리케이션 트랜잭션처럼 모델이 다음 단계를 바꾸면 안 되는 흐름에 알맞습니다.

# Wizard에서 Deterministic run_turn을 선택합니다.
uv run llamon orch settlement

CI에서 같은 scaffold를 반복 생성할 때만 선택값을 모두 적습니다.

# 비대화형 생성은 workflow와 자식 target을 고정합니다.
uv run llamon orch settlement \
  --workflow deterministic \
  --agent inventory=registry:inventory \
  --agent payment=registry:payment \
  --agent response=registry:settlement-response \
  --yes

생성된 run_turn에는 @managed_run_turn을 유지하세요. decorator가 durable 호출 원장, FinalResponse, output guardrail, WorkMemory 확정과 단 한 번의 최종 emit을 관리합니다.

순차 호출

app/orchestrator.py

from llamon_agent.orchestrator import managed_run_turn


@managed_run_turn(
    name="settlement",
    declared_aliases=("inventory", "payment", "response"),
)
async def run_turn(ctx):
    # 같은 alias를 다시 쓸 가능성이 있으므로 단계별 call_key를 고정합니다.
    stock = await ctx.call(
        "inventory",
        data=ctx.data,
        files=ctx.files,
        call_key="check-inventory",
    )
    if stock.is_input_required or stock.is_error or stock.is_unavailable:
        return stock

    # 다음 자식에는 검토한 필드만 새 payload로 투영합니다.
    payment = await ctx.call(
        "payment",
        data={"orderId": stock.data.get("orderId")},
        call_key="check-payment",
    )
    if payment.is_input_required or payment.is_error or payment.is_unavailable:
        return payment

    return await ctx.call(
        "response",
        data={
            "inStock": stock.data.get("inStock"),
            "paid": payment.data.get("paid"),
        },
        call_key="write-response",
    )

관리형 ctx.call()은 완료 결과를 durable 원장에 기록합니다. 바깥 WorkflowState 저장을 재시도하더라도 같은 invocation과 call_key의 자식을 다시 호출하지 않고 결과를 복구합니다. call_key는 자식 wire에 전달되지 않는 로컬 단계 식별자입니다.

같은 alias를 한 번만 호출하고 재개 분기에서도 순서가 바뀌지 않는다면 생략할 수 있습니다. 같은 alias를 여러 단계에서 호출하거나, 다음 턴에 앞 단계를 건너뛸 수 있다면 각 호출에 고유한 값을 지정하세요.

병렬 호출

독립된 조회는 ctx.gather()로 묶습니다. 일부 실패를 업무 결과에 반영하려면 return_exceptions=True로 결과를 모두 받은 뒤 직렬로 정리합니다.

from llamon_agent.orchestrator import AgentCallResult


async def collect_checks(ctx):
    checks = await ctx.gather(
        {
            "inventory": ctx.call(
                "inventory", data=ctx.data, call_key="parallel-inventory"
            ),
            "payment": ctx.call(
                "payment", data=ctx.data, call_key="parallel-payment"
            ),
        },
        return_exceptions=True,
    )

    # 예외와 제어 상태를 정상 결과로 오해하지 않도록 따로 표시합니다.
    completed = {
        name: value
        for name, value in checks.items()
        if isinstance(value, AgentCallResult)
        and not value.is_input_required
        and not value.is_error
        and not value.is_unavailable
    }
    incomplete = [name for name in checks if name not in completed]
    return completed, incomplete

기본 gather()는 시작된 호출을 모두 정리한 뒤 입력 순서상 첫 예외를 올립니다. 형제 호출의 외부 부수 효과가 즉시 취소된다고 가정하지 마세요. 병렬 실행 중에는 remember, reduce, append_message, emit처럼 공유 state를 바꾸지 말고, 결과를 받은 뒤 한 번에 반영합니다.

재진입 설계

Python의 실행 위치는 checkpoint에 저장되지 않습니다. 작업 자식이 input_required를 반환하면 다음 턴의 callback은 첫 줄부터 다시 실행됩니다. @managed_run_turn은 완료한 관리형 호출을 원장에서 복구하지만, 일반 Python 분기와 외부 부수 효과까지 되돌리지는 않습니다.

다단계 대기 흐름은 ctx.set_resume()과 ctx.get_resume()으로 다음 단계를 명시하세요. resume 값은 허용 목록에 매핑하고 사용자 입력을 alias나 함수 이름으로 직접 사용하지 않습니다.

@managed_run_turn(name="documents", declared_aliases=("verification",))
async def run_turn(ctx):
    resume = ctx.get_resume() or {}

    # 사용자가 보충 자료를 보낸 턴에는 대기 중이던 단계만 재개합니다.
    if resume.get("next_action") == "await_documents":
        result = await ctx.call(
            "verification",
            text=ctx.user_text,
            data=ctx.data,
            files=ctx.files,
            call_key="verification",
        )
        if not result.is_input_required:
            ctx.clear_resume()
        return result

    result = await ctx.call(
        "verification",
        data=ctx.data,
        files=ctx.files,
        call_key="verification",
    )
    if result.is_input_required:
        ctx.set_resume(next_action="await_documents")
    return result

재진입 전에 놓이는 외부 작업은 read-only이거나 멱등해야 합니다. 실행 여부가 불명확해질 수 있는 side effect는 자식이 orchestration ID를 멱등성 키로 사용하게 하거나 AgenticProgram의 승인된 durable 도구 경계로 옮기세요.

ctx.call_stream()은 관리형 호출 원장 대상이 아닙니다. 이미 공개한 chunk를 재개 때 회수할 수 없으므로 reviewer와 output guardrail을 모두 통과한 최종 text에만 사용합니다.

Agentic controller가 안전한 진행 상황을 직접 공개해야 할 때만 프로그램에 public_emissions = true를 선언합니다. controller JSON의 emit.kind="progress"는 일시적인 TaskStatusUpdateEvent, emit.kind="answer"는 최종 Task에도 누적되는 TaskArtifactUpdateEvent가 됩니다. 선언되지 않은 child raw chunk는 계속 internal이고, 두 emission 모두 설정된 output guardrail의 verified stream 경계를 통과한 뒤 전송됩니다. public_emissions와 output guardrail mode="stream"은 서로 독립된 opt-in입니다. guardrail이 없어도 progress는 언제나 일시적 status로만 나가며 최종 답변 text나 artifact metadata에 섞이지 않습니다.

결과와 예외

is_input_required, is_error, is_unavailable은 결과 상태입니다. 특히 is_unavailable은 upstream _pending_agent sentinel을 나타내는 준비 중 결과이지 네트워크 장애의 포괄 표현이 아닙니다. target 해석 실패와 전송·프로토콜 장애는 ChildUnresolvedError, ChildCallError 예외로 처리합니다.

세부 API는 OrchestratorContext API, 오류 공개 계약과 PostgreSQL 운영은 운영과 실패 복구을 참고하세요.