AI orchestration vs a durable task queue

from typing import Annotated

from zenml import pipeline, step, wait
from zenml.client import Client
from zenml.config.retry_config import StepRetryConfig

@step(retry=StepRetryConfig(max_retries=3, delay=10, backoff=2))
def research(topic: str) -> Annotated[str, "brief"]:
  return call_llm(f"Research: {topic}")

@step
def draft(brief: str) -> Annotated[str, "draft"]:
  return call_llm(f"Write a draft from:\n{brief}")

@step
def publish(text: str) -> None:
  send_to_cms(text)

@pipeline(dynamic=True)
def review_pipeline(topic: str):
  text = draft(research(topic))
  approved = wait(
      schema=bool,
      question="Publish this draft?",
      name="human_approval",
  )
  if approved:
      publish(text)

review_pipeline("AI orchestration")

# The run pauses at wait(). Resolve it in the dashboard, or:
#   zenml pipeline runs wait-conditions resolve --run <id> --interactive
#   zenml pipeline runs resume <id>
# Edit draft() and re-run: research comes from cache.
Client().get_artifact_version("draft").load()

from hatchet_sdk import Context, DurableContext, Hatchet, UserEventCondition
from pydantic import BaseModel

hatchet = Hatchet()

class ReviewInput(BaseModel):
  topic: str

review = hatchet.workflow(
  name="review", input_validator=ReviewInput, on_events=["review:create"]
)

@review.task(retries=3, backoff_factor=2)
def research(input: ReviewInput, ctx: Context) -> dict:
  return {"brief": call_llm(f"Research: {input.topic}")}

@review.task(parents=[research])
def draft(input: ReviewInput, ctx: Context) -> dict:
  brief = ctx.task_output(research)["brief"]
  return {"draft": call_llm(f"Write a draft from:\n{brief}")}

@review.durable_task(parents=[draft])
async def approve(input: ReviewInput, ctx: DurableContext) -> dict:
  # Consumes nothing while it waits; the worker slot can be freed.
  await ctx.aio_wait_for(
      "approval", UserEventCondition(event_key="review:approved")
  )
  return {"approved": True}

@review.task(parents=[approve])
def publish(input: ReviewInput, ctx: Context) -> dict:
  send_to_cms(ctx.task_output(draft)["draft"])
  return {}

# Workers pull from the queue; the engine owns retries, concurrency
# and rate limits. Task outputs live in the run's history, so
# persisting artifacts is your call.
worker = hatchet.worker("review-worker", workflows=[review])
worker.start()